Echtzeitdaten mit Änderungsstreams lesen

Mit Änderungsstreams für Firestore mit MongoDB-Kompatibilität können Anwendungen in Echtzeit auf Ä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.

Auf Änderungsstreams kann über die MongoDB-kompatiblen APIs und herkömmlichen MongoDB-Treiber zugegriffen werden. Die Implementierung von Änderungsstreams in Firestore mit MongoDB-Kompatibilität kann jeden Durchsatz von Schreib- und Lesevorgängen verarbeiten. Dies wird durch eine einzigartige Implementierung der automatischen Partitionierung bei Schreibvorgängen und der Leseparallelität ermöglicht. So können Sie Arbeitslasten mit hohem Durchsatz erstellen. Außerdem können Sie die Infrastruktur für die Migration und Datensynchronisierung zwischen Cloud Firestore und anderen Speicherlösungen verbessern.

Neben der Kompatibilität mit den MongoDB-Treibern, können Sie Cloud Firestore verwenden, um Änderungsstreams parallel zu lesen. So können Sie parallele Lese-Arbeitslasten mit hohem Durchsatz erstellen. Jeder Stream stellt eine gut verteilte Partition der Ergebnisse dar.

Änderungsstreams unterstützen die folgenden Funktionen:

  • Konfigurierbare Änderungsstreams mit Datenbank- oder Sammlungsumfang.
  • 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 und darf maximal 7 Tage betragen. Die Aufbewahrungsdauer kann nach der Erstellung nicht mehr geändert werden. Wenn Sie die Aufbewahrungsdauer ändern möchten, müssen Sie den Änderungsstream löschen und neu erstellen.
  • Änderungsereignisse vom Typ delete, insert, update und drop, die mit db.collection.watch() und db.watch() beobachtet werden können.
  • updateDescription.updatedFields enthält Aktualisierungsunterschiede.
  • Alle fullDocument- und fullDocumentBeforeChange-Optionen.
    • Vollständiges Dokument für Aktualisierungen suchen.
    • Vorheriges Image des Dokuments, bevor es ersetzt, aktualisiert oder gelöscht wurde.
    • Nachheriges Image des Dokuments, nachdem es ersetzt oder aktualisiert wurde.
    • Für Vorher- und Nachher-Images, die älter als eine Stunde sind, muss die Wiederherstellung zu einem bestimmten Zeitpunkt (Point-In-Time Recovery, PITR) aktiviert sein.
  • 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 Prinzipal die IAM-Berechtigungen datastore.schemas.create, datastore.schemas.delete und datastore.schemas.list.

Die Rolle Datastore-Index-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 in 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 , klicken Sie auf das Dreipunkt-Menü Weitere Aktionen und wählen Sie dann Änderungsstream erstellen aus.
  4. Geben Sie einen eindeutigen Namen, einen Umfang und eine Aufbewahrungsdauer für den Änderungsstream ein und klicken Sie dann auf Speichern.

Änderungsstreams ansehen

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

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

    Zur Seite „Datenbanken“

  2. Wählen Sie in 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 in 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.

Änderungsstream-Cursor öffnen oder fortsetzen

Die folgenden Beispiele zeigen, 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 die Methode watch in den MongoDB-Treibern, um einen neuen Änderungsstream-Cursor zu erstellen. Wenn Sie alle Änderungen an einer Datenbank beobachten möchten, erstellen Sie einen Änderungsstream mit Datenbankumfang und rufen Sie die Methode watch für das Objekt db auf.

let cursor = db.watch()

Wenn Sie einen Cursor mit Sammlungsumfang 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 für den 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 Optionen resumeAfter oder startAfter, um einen Änderungsstream fortzusetzen. Verwenden Sie ein Wiederaufnahmetoken, um zu bestimmen, an welcher Stelle im Änderungsprotokoll die Fortsetzung mit resumeAfter und startAfter erfolgen soll.

// 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())

So verwenden Sie startAfter:

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

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

Bei Bedarf können Sie Vorher- und Nachher-Images von Dokumenten in Änderungsereignisse vom Typ „Aktualisieren“ und „Löschen“ einbeziehen. Die Verfügbarkeit von Images hängt vom Fenster für die Wiederherstellung zu einem bestimmten Zeitpunkt (Point-In-Time Recovery, PITR) ab. Wenn Sie Dokument-Images lesen möchten, die älter als eine Stunde sind, müssen Sie PITR aktivieren.

Änderungsstreams nutzen das PITR-Fenster, um eine Ansicht des Dokuments vor und nach dem jeweiligen Änderungsereignis zu liefern. Standardmäßig enthalten Aktualisierungsereignisse das Feld updateDescription, das die Differenz der Felder darstellt, die durch den Aktualisierungsvorgang geändert wurden.

Wenn Sie die Vorher- und Nachher-Images in ein Änderungsereignis einbeziehen möchten, müssen Sie die Optionen fullDocumentBeforeChange und fullDocument in der Änderungs stream-Abfrage 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 mit dem Wert required eine serverseitige Fehlermeldung ausgegeben.

Alternativ zur Ausgabe eines Fehlers können Sie den Wert whenAvailable verwenden, um einen null-Wert zurückzugeben, wenn die Images nicht mehr verfügbar sind.

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

Aktuelles Image in Aktualisierungen einbeziehen

Standardmäßig enthalten Aktualisierungsereignisse das Feld updateDescription, das die Differenz der Felder darstellt, die durch den Aktualisierungsvorgang geändert wurden. Wenn Sie stattdessen die aktuellste Version des gesamten Dokuments suchen möchten, verwenden Sie den Wert updateLookup in der Option fullDocument.

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

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

Parallele Lesevorgänge

Um den Durchsatz zu erhöhen, können Sie mit der Option firestoreWorkerConfig eine Änderungsstream-Abfrage auf mehrere Worker aufteilen. 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 bei der Wiederherstellung aus einer Sicherung 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

In diesem 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, die wichtigsten Unterschiede sind:

  • In updateDescription werden die Felder truncatedArrays und disambiguatedPaths nicht ausgefüllt.
  • updateDescription.updatedFields stellen eine kanonische Differenz zwischen den Vorher- und Nachher-Images eines Dokuments dar, bevor und nachdem eine Mutation angewendet wurde.

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 mutieren.

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 aktualisiert der Vorgang nur das erste Array-Feld, überschreibt aber das gesamte Array.

Die Cloud Firestore Aktualisierungsdifferenz 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