Echtzeitdaten mit Änderungsstreams lesen

Mit Änderungsstreams für Firestore mit MongoDB-Kompatibilität können Anwendungen auf Echtzeitänderungen (Einfügungen, Aktualisierungen und Löschungen) zugreifen, die an einer Sammlung oder einer gesamten Datenbank vorgenommen wurden. In einem Änderungsstream werden Aktualisierungen nach Änderungszeit sortiert.

Änderungsstreams sind über die MongoDB-kompatiblen APIs und herkömmliche MongoDB-Treiber zugänglich. Die Implementierung von Änderungsstreams in Firestore mit MongoDB-Kompatibilität kann jeden Durchsatz von Schreib- und Lesevorgängen verarbeiten. Dazu wird eine einzigartige Implementierung der automatischen Partitionierung bei Schreibvorgängen und der Parallelität bei Lesevorgängen verwendet. So können Sie Arbeitslasten mit hohem Durchsatz erstellen. Sie können auch die Infrastruktur für die Migration und Datensynchronisierung zwischen Cloud Firestore und anderen Speicherlösungen verbessern.

Zusätzlich zur Kompatibilität mit den MongoDB-Treibern können Sie Cloud Firestore verwenden, um Änderungsstreams parallel zu lesen. So können Sie parallele Lesearbeitslasten mit hohem Durchsatz erstellen. Jeder Stream stellt eine gut verteilte Partition von Ergebnissen dar.

Änderungsstreams unterstützen die folgenden Funktionen:

  • Konfigurierbare Änderungsstreams mit Datenbank- oder Sammlungsbereich.
  • Eine Aufbewahrungsdauer für einen Änderungsstream, die bei der Erstellung angegeben wird. Die Standardaufbewahrungsdauer beträgt 7 Tage und die Mindestaufbewahrungsdauer 1 Tag. Die Aufbewahrungsdauer muss ein Vielfaches von 1 Tag sein, bis zu maximal 7 Tagen. Die Aufbewahrungsdauer kann nach dem Erstellen nicht mehr geändert werden. Wenn Sie die Aufbewahrungsdauer ändern möchten, müssen Sie den Änderungsstream löschen und neu erstellen.
  • delete-, insert-, update- und drop-Änderungsereignisse, die mit db.collection.watch() und db.watch() beobachtet werden können.
  • updateDescription.updatedFields enthält Update-Diffs.
  • Alle fullDocument- und fullDocumentBeforeChange-Optionen.
    • Das vollständige Dokument wird auf Aktualisierungen geprüft.
    • Pre-Image des Dokuments, bevor es ersetzt, aktualisiert oder gelöscht wurde.
    • Post-Image des Dokuments, nachdem es ersetzt oder aktualisiert wurde.
    • Für Pre- und Post-Images, die älter als eine Stunde sind, muss die Wiederherstellung zu einem bestimmten Zeitpunkt (Point-in-Time Recovery, PITR) aktiviert werden.
  • Alle Optionen zum Fortsetzen, einschließlich resumeAfter und startAfter.
  • Wenn Sie watch() verwenden, um Änderungen zu beobachten, können Sie Aggregationsphasen wie $addFields, $match, $project, $replaceRoot, $replaceWith, $set und $unset verketten.

Änderungsstreams konfigurieren

Verwenden Sie die Google Cloud Console, um vorhandene Änderungsstreams für eine Datenbank zu erstellen, zu löschen oder anzusehen.

Rollen und Berechtigungen

Zum Erstellen, Löschen und Auflisten von Änderungsstreams benötigt ein Hauptkonto die IAM-Berechtigungen (Identity and Access Management) datastore.schemas.create, datastore.schemas.delete und datastore.schemas.list.

Die Rolle Datenspeicherindex-Administrator (roles/datastore.indexAdmin) gewährt beispielsweise diese Berechtigungen.

Änderungsstream erstellen

Bevor Sie einen entsprechenden Änderungsstream-Cursor öffnen können, müssen Sie einen Änderungsstream erstellen. Die automatische Aktivierung von Änderungsstreams beim Erstellen von Sammlungen oder Datenbanken wird nicht unterstützt.

Verwenden Sie die Google Cloud Console, um einen Änderungsstream zu erstellen.

  1. Rufen Sie in der Google Cloud Console die Seite Datenbanken auf.

    Zur Seite „Datenbanken“

  2. Wählen Sie aus der Liste eine Firestore-Datenbank mit MongoDB-Kompatibilität aus. Der Bereich Firestore Studio wird geöffnet.
  3. Suchen Sie im Bereich Explorer den Knoten Änderungsstreams, klicken Sie auf Weitere Aktionen und wählen Sie dann Änderungsstream erstellen aus.
  4. Geben Sie einen eindeutigen Namen, einen Bereich und eine Aufbewahrungsdauer für den Änderungsstream ein und klicken Sie dann auf Speichern.

Änderungsstreams ansehen

In der Google Cloud Console können Sie sich Details zu Änderungsstreams ansehen.

  1. Rufen Sie in der Google Cloud Console die Seite Datenbanken auf.

    Zur Seite „Datenbanken“

  2. Wählen Sie aus der Liste eine Firestore-Datenbank mit MongoDB-Kompatibilität aus. Der Bereich Firestore Studio wird geöffnet.
  3. Suchen Sie im Bereich Explorer nach dem Knoten Änderungsstreams.
  4. Klicken Sie auf Knoten ein-/ausblenden, um den Knoten zu öffnen oder zu schließen.

Änderungsstream löschen

Verwenden Sie die Google Cloud Console, um einen Änderungsstream zu löschen.

  1. Rufen Sie in der Google Cloud Console die Seite Datenbanken auf.

    Zur Seite „Datenbanken“

  2. Wählen Sie aus der Liste eine Firestore-Datenbank mit MongoDB-Kompatibilität aus. Der Bereich Firestore Studio wird geöffnet.
  3. Suchen Sie im Bereich Explorer nach dem Knoten Änderungsstreams.
  4. Klicken Sie auf Knoten ein-/ausblenden, um den Knoten zu öffnen oder zu schließen.
  5. Suchen Sie im Explorer nach dem Änderungsstream, den Sie löschen möchten.
  6. Klicken Sie auf Weitere Aktionen und wählen Sie dann Änderungsstream löschen aus.
  7. Geben Sie im Dialogfeld den Namen des Änderungsstreams ein, um das Löschen zu bestätigen, und klicken Sie dann auf Löschen.

Cursor für Änderungsstream öffnen oder fortsetzen

In den folgenden Beispielen wird gezeigt, wie Sie einen Änderungsstream-Cursor erstellen, fortsetzen und konfigurieren.

Bevor Sie einen Änderungsstream-Cursor erstellen können, müssen Sie explizit einen Änderungsstream für die Datenbank oder Sammlung erstellen.

Änderungsstream-Cursor erstellen

Verwenden Sie zum Erstellen eines neuen Änderungsstream-Cursors die Methode watch in den MongoDB-Treibern. Wenn Sie alle Änderungen in einer Datenbank überwachen möchten, erstellen Sie einen Änderungsstream auf Datenbankebene und rufen Sie die Methode watch für das Objekt db auf.

let cursor = db.watch()

Wenn Sie einen Cursor für eine Sammlung erstellen möchten, müssen Sie zuerst einen Änderungsstream für diese Sammlung erstellen. Rufen Sie dann die Methode watch für die entsprechende Sammlung auf.

let cursor = db.my_collection.watch()

Nachdem Sie einen Änderungsstream-Cursor erstellt haben, können Sie mit dem Streaming beginnen. Wenn Sie beispielsweise ein Dokument einfügen und tryNext auf dem Cursor aufrufen, wird die Änderung im Änderungsstream angezeigt.

let doc = db.my_collection.insertOne({value: "hello world"})
console.log(cursor.tryNext())

Wenn Sie das Dokument aktualisieren und löschen, werden diese Änderungen im Änderungsstream angezeigt:

db.my_collection.updateOne({"_id": doc.insertedId}, {$set: {value: "hello world!"}})
db.my_collection.deleteOne({"_id": doc.insertedId}})

// Prints the update event
console.log(cursor.tryNext())

// Prints the delete event
console.log(cursor.tryNext())

Änderungsstream fortsetzen

Verwenden Sie die Option resumeAfter oder startAfter, um einen Änderungsstream fortzusetzen. Verwenden Sie ein Fortsetzungstoken, um zu bestimmen, wo im Änderungsprotokoll resumeAfter und startAfter fortgesetzt werden sollen.

// Create a cursor and add one event to the change stream.
let cursor = db.my_collection.watch();
db.my_collection.insertOne({value: "hello world"});
let event = cursor.tryNext();

// Get the resume token from the event.
let resumeToken = event._id;

// Add a new event to the change stream.
db.my_collection.insertOne({value: "foobar"});

// Create a new cursor by using the resume token as a starting point.
let newCursor = db.my_collection.watch({resumeAfter: resumeToken})

// Log the change event containing the "foobar" value.
console.log(newCursor.tryNext())

Mithilfe von startAfter:

// Start after the resume token.
let startAfterCursor = db.my_collection.watch({startAfter: resumeToken})

Vorher- und Nachher-Bilder in Aktualisierungen und Löschvorgänge einbeziehen

Bei Bedarf können Sie Vorher- und Nachher-Bilder von Dokumenten in Änderungsereignisse für Aktualisierungen und Löschungen einfügen. Die Verfügbarkeit von Bildern hängt vom Fenster für die Wiederherstellung zu einem bestimmten Zeitpunkt ab. Wenn Sie Dokumentbilder lesen möchten, die älter als eine Stunde sind, müssen Sie die Wiederherstellung zu einem bestimmten Zeitpunkt aktivieren.

Änderungsstreams nutzen das PITR-Fenster, um eine Ansicht des Dokuments vor und nach dem jeweiligen Änderungsereignis bereitzustellen. Standardmäßig enthalten Aktualisierungsereignisse ein updateDescription-Feld, das das Delta der durch den Aktualisierungsvorgang geänderten Felder ist.

Wenn Sie die Vorher- und Nachher-Bilder in ein Änderungsereignis aufnehmen möchten, müssen Sie die Optionen fullDocumentBeforeChange und fullDocument in der Änderungsstreamabfrage angeben.

let cursor = db.my_collection.watch({
  "fullDocument": "required",
  "fullDocumentBeforeChange": "required"
})

Wenn die Abfrage versucht, ein Dokument außerhalb des PITR-Aufbewahrungsfensters zu lesen, oder wenn PITR nicht aktiviert ist, wird für den Wert required eine serverseitige Fehlermeldung ausgegeben.

Als Alternative zum Auslösen eines Fehlers können Sie den Wert whenAvailable verwenden, um den Wert null zurückzugeben, wenn die Bilder nicht mehr verfügbar sind.

let cursor = db.my_collection.watch({
  "fullDocument": "whenAvailable",
  "fullDocumentBeforeChange": "whenAvailable"
})

Aktuelles Bild in Updates einbeziehen

Standardmäßig enthalten Aktualisierungsereignisse ein updateDescription-Feld, das das Delta der durch den Aktualisierungsvorgang geänderten Felder ist. Wenn Sie stattdessen die aktuelle Version des gesamten Dokuments abrufen möchten, verwenden Sie den Wert updateLookup in der Option fullDocument.

Für diese Funktion ist kein PITR erforderlich. Es wird eine Suche nach dem Dokument durchgeführt.

let cursor = db.my_collection.watch({
  "fullDocument": "updateLookup",
})

Parallele Lesevorgänge

Wenn Sie den Durchsatz erhöhen möchten, können Sie die Option firestoreWorkerConfig verwenden, um eine Änderungsstream-Abfrage auf mehrere Worker aufzuteilen. Jeder Worker ist für die Bereitstellung der Änderungen für eine bestimmte Gruppe von Dokumenten verantwortlich. Sie müssen einen parallelen Cursor über eine runCommand- oder aggregate-Abfrage erstellen.

Sie können einen Änderungsstream beispielsweise so auf drei Worker verteilen:

let cursor1 = db.my_collection.aggregate([{
    "$changeStream": {
        "firestoreWorkerConfig": {numWorkers: 3, workerId: 0 }}
  }]);

let cursor2 = db.my_collection.aggregate([{
    "$changeStream": {
        "firestoreWorkerConfig": {numWorkers: 3, workerId: 1 }}
  }]);

let cursor3 = db.my_collection.aggregate([{
    "$changeStream": {
        "firestoreWorkerConfig": {numWorkers: 3, workerId: 2 }}
  }]);

Änderungsstreams und Sicherungen

Weder die Konfiguration des Änderungsstreams noch die Daten des Änderungsstreams sind in Sicherungswiederherstellungsvorgängen verfügbar. Wenn Sie eine Datenbank mit Änderungsstreams wiederherstellen, müssen Sie diese Änderungsstreams in der Zieldatenbank neu erstellen, um Cursor für diese Datenbank zu öffnen.

Abrechnung

Verhaltensunterschiede

Im folgenden Abschnitt werden die Unterschiede bei Änderungsstreams zwischen Firestore mit MongoDB-Kompatibilität und MongoDB beschrieben.

updateDescription

updateDescription ist ein Dokument in einem update-Ereignis, das die Felder beschreibt, die durch den Aktualisierungsvorgang aktualisiert oder entfernt wurden. In Cloud Firestore gibt es folgende wichtige Unterschiede:

  • In updateDescription sind die Felder truncatedArrays und disambiguatedPaths nicht ausgefüllt.
  • updateDescription.updatedFields stellt einen kanonischen Unterschied zwischen den Bildern eines Dokuments vor und nach der Anwendung einer Mutation dar.

Betrachten Sie den folgenden Ausgangszustand eines Dokuments:

db.my_collection.insertOne({
  _id: 1,
  root: {
    array: [{a: 1}, {b: 2}, {c: 3}]
  }
})

Szenario 1: Nur das erste Element des Arrays wird geändert.

In diesem Szenario entspricht das Verhalten von Cloud Firestore dem von MongoDB.

db.my_collection.updateOne(
  {_id: 1},
  {'$set': {"root.array.0.a": 100}}
)

{
  updatedFields: {"root.array.0.a": 100},
  removedFields: []
}

Szenario 2: Mit einem ganzen Array überschreiben

In diesem Szenario wird nur das erste Arrayfeld aktualisiert, aber das gesamte Array überschrieben.

Das Update-Diff Cloud Firestore unterscheidet nicht zwischen diesen beiden Szenarien und gibt für beide dasselbe updateDescription.updatedFields zurück:

db.my_collection.updateOne(
  {_id: 1},
  {'$set': {"root.array": [{a: 100}, {b: 2}, {c: 3}]}}
)

// In other implementations, updatedFields reflects the mutation itself
{
  updatedFields: {
    "root.array": [{a: 100}, {b: 2}, {c: 3}]
  },
  removedFields: []
}

// Firestore updatedFields is the diff between the before and after versions of the document
{
  updatedFields: {"root.array.0.a": 100},
  removedFields: []
}

Nächste Schritte