Ereignisse und Streams

Datastream organisiert replizierte Daten in einer dreistufigen Hierarchie aus Streams, Objekten und Ereignissen. Ein Stream verbindet eine Datenquelle mit einem Ziel, während ein Objekt eine bestimmte Quellentität in diesem Stream darstellt, z. B. eine Datenbanktabelle. Ein Ereignis erfasst eine einzelne Datenänderung von einem Objekt, z. B. einen Datenbankeintrag.

Jedes Ereignis enthält Daten und Metadaten, die den Quellstream und das Objekt identifizieren. Mit diesen Metadaten können Sie den Ursprung von Änderungen nachverfolgen, doppelte Datensätze im Ziel entfernen und die Reihenfolge für nachgelagerte Nutzer beibehalten.

Über Ereignisse

Jedes Ereignis besteht aus drei Datentypen:

  • Ereignisdaten: Dies ist die Änderung der Daten selbst aus dem Objekt, das von der Streamquelle stammt. Jedes Ereignis enthält die gesamte geänderte Zeile.
  • Allgemeine Metadaten: Diese Metadaten werden für jedes von Datastream generierte Ereignis angezeigt, das für Aktionen wie das Entfernen doppelter Daten im Ziel verwendet wird.
  • Quellspezifische Metadaten: Diese Metadaten werden für jedes Ereignis angezeigt, das von einer bestimmten Streamquelle generiert wird. Diese Metadaten variieren je nach Quelle.

Ereignisdaten

Ereignisdaten sind die Nutzlast jeder Änderung aus einem bestimmten Objekt, das aus einer Streamquelle stammt.

Ereignisse liegen entweder im Avro- oder im JSON-Format vor.

Wenn Sie mit dem Avro-Format arbeiten, enthält das Ereignis für jede Spalte den Spaltenindex und den Wert. Anhand des Spaltenindex können der Spaltenname und der vereinheitlichte Typ aus dem Schema im Avro-Header abgerufen werden.

Wenn Sie mit dem JSON-Format arbeiten, enthält das Ereignis für jede Spalte den Spaltennamen und den Wert.

Ereignismetadaten können verwendet werden, um Informationen zum Ursprung des Ereignisses zu erfassen und doppelte Daten am Ziel zu entfernen und Ereignisse nach dem nachgelagerten Nutzer zu sortieren.

In den folgenden Tabellen werden die Felder und Datentypen für allgemeine und quellenspezifische Ereignismetadaten aufgelistet und beschrieben.

Generische Metadaten

Diese Metadaten sind für alle Arten von Streams einheitlich.

Feld Avro-Typ JSON-Typ Beschreibung
stream_name String String Der eindeutige Streamname, der bei der Erstellung definiert wurde.
read_method String String

Gibt an, ob die Daten aus der Quelle mit einer CDC-Methode (Change Data Capture) als Teil des verlaufsbezogenen Backfills oder als Teil einer Ergänzungsaufgabe gelesen wurden, die beim Rollback einer Transaktion während der CDC-Replikation erstellt wird.

Zulässige Werte:

  • cdc
  • backfill
  • oracle-cdc-logminer
  • oracle-backfill
  • oracle-supplementation
  • mysql-cdc-binlog
  • mysql-backfill-incremental
  • mysql-backfill-fulldump
  • postgres-cdc-wal
  • postgresql-backfill
  • sqlserver-cdc
  • sqlserver-backfill
  • salesforce-cdc
  • salesforce-backfill
object String String Der Name, mit dem verschiedene Ereignistypen gruppiert werden, in der Regel der Name der Tabelle oder des Objekts in der Quelle.
schema_key String String Eine eindeutige Kennung für das einheitliche Schema des Ereignisses.
uuid String String Eine eindeutige Kennung für das von Datastream generierte Ereignis.
read_timestamp timestamp-millis String Der Zeitstempel (UTC), wenn der Datensatz von Datastream gelesen wurde (der Epochenzeitstempel in Millisekunden).
source_timestamp timestamp-millis String Der Zeitstempel (UTC), wenn sich der Eintrag in der Quelle geändert hat (der Epochenzeitstempel in Millisekunden).
sort_keys {"type": "array", "items": ["string", "long"]} array Ein Array von Werten, mit denen die Ereignisse in der Reihenfolge sortiert werden können, in der sie aufgetreten sind.

Quellspezifische Metadaten

Diese Metadaten sind CDC- und Backfill-Ereignissen aus einer Quelldatenbank zugeordnet. Wählen Sie im Drop-down-Menü unten eine Quelle aus, um diese Metadaten aufzurufen.

Quelle Feld Avro-Typ JSON-Typ Beschreibung
MySQL log_file String String Die Logdatei, aus der Datastream bei der CDC-Replikation Ereignisse abruft, wenn die binlogbasierte Replikation verwendet wird.
MySQL log_position long long Die Logposition (Offset) im binären MySQL-Log bei Verwendung der binlogbasierten Replikation.
MySQL binlog_file String String Die binäre Logdatei, aus der Datastream bei der CDC-Replikation Ereignisse abruft, wenn die GTID-basierte Replikation verwendet wird.
MySQL binlog_position long long Die Position (Offset) im binären MySQL-Log bei Verwendung der GTID-basierten Replikation.
MySQL gtid String String Die globale Transaktions-ID (GTID), die die Transaktion für das Ereignis bei Verwendung der GTID-basierten Replikation eindeutig identifiziert.
MySQL datastream_master_server_uuid String String Die UUID der MySQL-Quellserverinstanz, aus der Datastream Ereignisse liest, wenn die GTID-basierte Replikation verwendet wird.
MySQL datastream_master_server_id long long Die Server-ID der MySQL-Quellserverinstanz, aus der Datastream Ereignisse liest, wenn die GTID-basierte Replikation verwendet wird.
MySQL primary_keys String-Array String-Array Die Liste mit einem oder mehreren Spaltennamen, aus denen der Primärschlüssel der Tabellen besteht. Wenn die Tabelle keinen Primärschlüssel hat, ist dieses Feld leer.
MySQL is_deleted Boolesch Boolesch
  • Der Wert true gibt an, dass die Zeile in der Quelle gelöscht wurde.
  • Der Wert false gibt an, dass die Zeile nicht gelöscht wurde.
MySQL database String String Die mit dem Ereignis verknüpfte Datenbank.
MySQL table String String Die mit dem Ereignis verknüpfte Tabelle.
MySQL change_type String String

Die Art der Änderung (INSERT, UPDATE-INSERT, UPDATE-DELETE und DELETE), die das Ereignis darstellt.
Oracle log_file String String Die Logdatei, aus der Datastream Ereignisse bei der CDC-Replikation abruft.
Oracle scn long long Die Logposition (Offset) im Oracle-Transaktionslog.
Oracle row_id String String row_id von Oracle.
Oracle is_deleted Boolesch Boolesch
  • Der Wert true gibt an, dass die Zeile in der Quelle gelöscht wurde.
  • Der Wert false gibt an, dass die Zeile nicht gelöscht wurde.
Oracle database String String Die mit dem Ereignis verknüpfte Datenbank.
Oracle schema String String Das Schema, das der Tabelle aus dem Ereignis zugeordnet ist.
Oracle table String String Die mit dem Ereignis verknüpfte Tabelle.
Oracle change_type String String

Die Art der Änderung (INSERT, UPDATE-INSERT, UPDATE-DELETE und DELETE), die das Ereignis darstellt.
Oracle tx_id String String Die Transaktions-ID, zu der das Ereignis gehört.
Oracle rs_id String String Die ID des Datensatzes. Die Kombination aus rs_id und ssn identifiziert eine Zeile in V$LOGMNR_CONTENTS eindeutig. rs_id identifiziert den Redo-Eintrag, der die Zeile generiert hat, eindeutig.
Oracle ssn long long Eine SQL-Sequenznummer. Diese Nummer wird mit rs_id verwendet und identifiziert eine Zeile in V$LOGMNR_CONTENTS eindeutig.
PostgreSQL schema String String Das Schema, das der Tabelle aus dem Ereignis zugeordnet ist.
PostgreSQL table String String Die mit dem Ereignis verknüpfte Tabelle.
PostgreSQL is_deleted boolean Boolesch
  • Der Wert true gibt an, dass die Zeile in der Quelle gelöscht wurde.
  • Der Wert false gibt an, dass die Zeile nicht gelöscht wurde.
PostgreSQL change_type String String Die Art der Änderung (INSERT, UPDATE, DELETE), die das Ereignis darstellt.
PostgreSQL tx_id String String Die Transaktions-ID, zu der das Ereignis gehört.
PostgreSQL lsn String String Die Logsequenznummer für den aktuellen Eintrag.
PostgreSQL primary_keys String-Array String-Array Die Liste mit einem oder mehreren Spaltennamen, aus denen der Primärschlüssel der Tabellen besteht. Wenn die Tabelle keinen Primärschlüssel hat, ist dieses Feld leer.
SQL Server table String String Die mit dem Ereignis verknüpfte Tabelle.
SQL Server database long long Die mit dem Ereignis verknüpfte Datenbank.
SQL Server schema String-Array String-Array Das Schema, das der Tabelle aus dem Ereignis zugeordnet ist.
SQL Server is_deleted boolean Boolesch
  • Der Wert true gibt an, dass die Zeile in der Quelle gelöscht wurde.
  • Der Wert false gibt an, dass die Zeile nicht gelöscht wurde.
SQL Server lsn String String Die Logsequenznummer für das Ereignis.
SQL Server tx_id String String Die Transaktions-ID, zu der das Ereignis gehört.
SQL Server physical_location Ganzzahl-Array Ganzzahl-Array Der physische Speicherort des Logeintrags, beschrieben durch drei Ganzzahlen: Datei-ID, Seiten-ID und Slot-ID des Eintrags.
SQL Server replication_index String-Array String-Array Die Liste der Spaltennamen eines Index, mit denen eine Zeile in der Tabelle eindeutig identifiziert werden kann.
SQL Server change_type String String

Die Art der Änderung (INSERT, UPDATE, DELETE), die das Ereignis darstellt.
Spanner commit_timestamp timestamp-micros String Der Commit-Zeitstempel des Ereignisses.
Spanner snapshot boolean boolean Ob das Ereignis ein Backfill-Snapshot-Ereignis ist.
Spanner project_id String String Die Spanner-Projekt-ID.
Spanner instance_id String String Die Cloud Spanner-Instanz-ID.
Spanner database_id String String Die Spanner-Datenbank-ID.
Spanner schema String String Das Schema, das der Tabelle aus dem Ereignis zugeordnet ist. Leer, wenn sich die Tabelle nicht in einem benannten Schema befindet.
Spanner table String String Die Cloud Spanner-Tabelle.
Spanner change_stream_name String String Der Cloud Spanner-Änderungsstream.
Spanner server_transaction_id String String Die Transaktions-ID, die mit dem Ereignis verknüpft ist.
Spanner record_sequence long oder string long oder string Die mit dem Ereignis verknüpfte Datensatzsequenz. Bei Änderungsstreams, die mit dem Partitionsmodus IMMUTABLE_KEY_RANGE erstellt wurden, ist der Wert eine numerische Sequenznummer vom Typ long. Bei Änderungsstreams, die mit dem Partitionsmodus MUTABLE_KEY_RANGE erstellt wurden, ist der Wert der String record_sequence aus dem Datenänderungsdatensatz, z. B. 1cac8f3f38eeaf53-00000000.
Spanner mod_number long long Die Änderungsnummer, die dem Ereignis zugeordnet ist.
Spanner transaction_tag String String Das Tag für die zugehörige Transaktion.
Spanner number_of_records_in_transaction long long Die Anzahl der Datensätze in der zugehörigen Transaktion.
Spanner value_capture_type String String Der Werterfassungstyp des Änderungsstreams. Mögliche Werte sind NEW_ROW und NEW_ROW_AND_OLD_VALUES.
Spanner change_type String String Die Art der Änderung (INSERT, UPDATE oder DELETE), die das Ereignis darstellt.
Spanner primary_keys String-Array String-Array Die Liste der Primärschlüsselspalten für die Tabelle.
Spanner is_deleted boolean boolean Gibt an, ob die Änderung ein DELETE-Ereignis darstellt.
Salesforce object_name String String

Der Name des Salesforce-Objekts, das mit dem Ereignis verknüpft ist.
Salesforce domain String String

Der Name der Domain, die mit dem Ereignis verknüpft ist.
Salesforce is_deleted boolean Boolesch
  • Der Wert true gibt an, dass die Zeile in der Quelle gelöscht wurde.
  • Der Wert false gibt an, dass die Zeile nicht gelöscht wurde.
Salesforce change_type String String

Die Art der Änderung (INSERT, UPDATE, DELETE), die das Ereignis darstellt.
Salesforce primary_keys String-Array String-Array Die Liste der Spaltennamen, die den Primärschlüssel der Tabelle bilden. Wenn die Tabelle keinen Primärschlüssel hat, ist dieses Feld leer.
MongoDB database String String Die mit dem Ereignis verknüpfte Datenbank.
MongoDB collection String String Die mit dem Ereignis verknüpfte Sammlung. Sammlungen entsprechen Tabellen in relationalen Datenbanken.
MongoDB change_type String String Die Art der Änderung (CREATE, UPDATE und DELETE), die das Ereignis darstellt.
MongoDB is_deleted boolean Boolesch
  • Der Wert true gibt an, dass die Zeile in der Quelle gelöscht wurde.
  • Der Wert false gibt an, dass die Zeile nicht gelöscht wurde.
MongoDB primary_keys String-Array String-Array Das Feld _id, das als Primärschlüssel für jedes Dokument in einer Sammlung dient.

Beispiel für einen Ereignisfluss

Dieser Ablauf veranschaulicht die Ereignisse, die von drei aufeinanderfolgenden Vorgängen generiert werden: INSERT, UPDATE und DELETE in einer einzelnen Zeile in einer SAMPLE-Tabelle für eine Quelldatenbank.

UHRZEIT THIS_IS_MY_PK (int) FIELD1 (nchar, Nullwerte zulässig) FIELD2 (nchar, keine Nullwerte)>
0 1231535353 foo TLV
1 1231535353 NULL TLV

INSERT (T0)

Die Nachrichtennutzlast besteht aus der gesamten neuen Zeile.

{
  "stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
  "read_method": "oracle-cdc-logminer",
  "object": "SAMPLE.TBL",
  "uuid": "d7989206-380f-0e81-8056-240501101100",
  "read_timestamp": "2019-11-07T07:37:16.808Z",
  "source_timestamp": "2019-11-07T02:15:39",  
  "source_metadata": {
    "log_file": ""
    "scn": 15869116216871,
    "row_id": "AAAPwRAALAAMzMBABD",
    "is_deleted": false,
    "database": "DB1",
    "schema": "ROOT",
    "table": "SAMPLE"
    "change_type": "INSERT",
    "tx_id": 
    "rs_id": "0x0073c9.000a4e4c.01d0",
    "ssn": 67,
  },
  "payload": {
    "THIS_IS_MY_PK": "1231535353",
    "FIELD1": "foo",
    "FIELD2": "TLV",
  }
}

UPDATE (T1)

Die Nachrichtennutzlast besteht aus der gesamten neuen Zeile. Vorherige Werte sind nicht enthalten.

{
  "stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
  "read_method": "oracle-cdc-logminer",
  "object": "SAMPLE.TBL",
  "uuid": "e6067366-1efc-0a10-a084-0d8701101101",
  "read_timestamp": "2019-11-07T07:37:18.808Z",
  "source_timestamp": "2019-11-07T02:17:39",  
  "source_metadata": {
    "log_file": 
    "scn": 15869150473224,
    "row_id": "AAAGYPAATAAPIC5AAB",
    "is_deleted": false,
    "database":
    "schema": "ROOT",
    "table": "SAMPLE"
    "change_type": "UPDATE",
    "tx_id":
    "rs_id": "0x006cf4.00056b26.0010",
    "ssn": 0,
  },
  "payload": {
    "THIS_IS_MY_PK": "1231535353",
    "FIELD1": null,
    "FIELD2": "TLV",
  }
}

DELETE (T2)

Die Nachrichtennutzlast besteht aus der gesamten neuen Zeile.

{
  "stream_name": "projects/myProj/locations/myLoc/streams/Oracle-to-Source",
  "read_method": "oracle-cdc-logminer",
  "object": "SAMPLE.TBL",
  "uuid": "c504f4bc-0ffc-4a1a-84df-6aba382fa651",
  "read_timestamp": "2019-11-07T07:37:20.808Z",
  "source_timestamp": "2019-11-07T02:19:39",
  "source_metadata": {
    "log_file": 
    "scn": 158691504732555,
    "row_id": "AAAGYPAATAAPIC5AAC",
    "is_deleted": true,
    "database":
    "schema": "ROOT",
    "table": "SAMPLE"
    "change_type": "DELETE",
    "tx_id":
    "rs_id": "0x006cf4.00056b26.0011",
    "ssn": 0,
  },
  "payload": {
    "THIS_IS_MY_PK": "1231535353",
    "FIELD1": null,
    "FIELD2": "TLV",
  }
}

Sortierung und Konsistenz

In diesem Abschnitt wird beschrieben, wie Datastream die Reihenfolge und Konsistenz handhabt.

Bestellung

Datastream garantiert nicht, dass Ereignisse in der richtigen Reihenfolge bereitgestellt werden. Jedes Ereignis enthält jedoch die vollständige Datenzeile, den Zeitstempel, zu dem die Daten in die Quelle geschrieben wurden, und Metadaten zur Reihenfolge.

Bei Ereignissen, die Teil eines ersten Backfills sind, ist das Feld read_method auf backfill gesetzt oder endet mit -backfill, -fulldump oder -incremental. Backfill-Ereignisse werden absichtlich in falscher Reihenfolge generiert und können in beliebiger Reihenfolge genutzt werden.

Bei Ereignissen, die Teil einer laufenden Change Data Capture-Replikation (CDC) sind, ist das Feld read_method auf cdc oder einen quellenspezifischen CDC-Wert wie mysql-cdc-binlog, oracle-cdc-logminer, postgres-cdc-wal, sqlserver-cdc oder salesforce-cdc gesetzt. Wie Sie laufende Replikationsereignisse über alle Quellen hinweg sortieren, hängt von Ihrem Ziel ab:

Ziel Beschreibung
BigQuery

Wie Ereignisse in BigQuery sortiert werden, hängt vom Schreibmodus ab, den Sie für Ihren Stream konfigurieren:

  • Zusammenführungs-Schreibmodus: BigQuery führt Ereignisse, die nicht in der richtigen Reihenfolge sind, automatisch mithilfe der Ereignismetadaten und einer internen Änderungssequenznummer (CSN) in der richtigen Reihenfolge zusammen. Tabellen ohne Primärschlüssel werden im Modus „Nur anhängen“ geschrieben.
  • Schreibmodus „Nur anhängen“, einschließlich Apache Iceberg-Tabellen: Datastream hängt Änderungsereignisse an die Zieltabelle an und füllt die Spalte datastream_metadata STRUCT aus. Wenn Sie die Reihenfolge bestimmen möchten, in der Ereignisse in der Quelle aufgetreten sind, sortieren Sie die Ereignisse nach dem Feld datastream_metadata.change_sequence_number oder dem Arrayfeld datastream_metadata.sort_keys.
Cloud Storage

In Cloud Storage können Ereignisse aus demselben Zeitraum mehrere Dateien umfassen. Wenn Sie die Reihenfolge bestimmen möchten, in der Ereignisse an der Quelle aufgetreten sind, sortieren Sie die Ereignisse mithilfe des Array-Felds sort_keys in den generischen Metadaten.

Konsistenz

Datastream sorgt dafür, dass die Daten aus der Quelldatenbank mindestens einmal an das Ziel geliefert werden. Es wird kein Ereignis verpasst, aber es besteht die Möglichkeit, dass Ereignisse im Stream doppelt vorkommen. Das Zeitfenster für doppelte Ereignisse sollte im Minutenbereich liegen. Die UUID (Universally Unique Identifier) des Ereignisses in den Ereignismetadaten kann verwendet werden, um Duplikate zu erkennen.

Verlassen Sie sich jedoch nicht ausschließlich auf die UUID, um Ereignisse nachgelagert zu deduplizieren. Wenn eine Pipeline unterbrochen wird, liest Datastream das Quelllog ab dem letzten gespeicherten Prüfpunkt noch einmal, um die Übermittlung sicherzustellen. Da Datastream beim Lesen eines Ereignisses dynamisch UUIDs generiert, führt diese Art von Wiederholung zu neuen UUIDs für genau dieselben Quelldaten. Außerdem können bei Bulk-Vorgängen in der Quelldatenbank mehrere unterschiedliche Zeilen dieselbe Änderungssequenznummer (CSN) haben.

Wenn Sie Daten zuverlässig deduplizieren möchten, partitionieren Sie die Ereignisse nach den Primärschlüsseln der Quelltabelle und gruppieren Sie sie mithilfe der Änderungssequenznummer (CSN) (Feld change_sequence_number). So wird sichergestellt, dass Wiederholungen desselben Ereignisses korrekt zusammengefasst werden, während legitime, unterschiedliche Zeilenaktualisierungen aus Bulk-Transaktionen erhalten bleiben.

Wenn Datenbank-Logdateien nicht übernommene Transaktionen enthalten und Transaktionen zurückgesetzt werden, wird dies in den Logdateien als „umgekehrte“ DML-Vorgänge (Datenbearbeitungssprache) dargestellt. Ein zurückgesetzter INSERT-Vorgang hat beispielsweise einen entsprechenden DELETE-Vorgang. Datastream liest diese Vorgänge aus den Logdateien.

Streams

Jeder Stream hat Metadaten, die sowohl den Stream als auch die Quelle beschreiben, aus der Daten abgerufen werden. Diese Metadaten enthalten Informationen wie den Streamnamen sowie die Quell- und Zielverbindungsprofile.

Die vollständige Definition des Stream-Objekts finden Sie in der Dokumentation API-Referenz.

Streamzustand und -status

Ein Stream kann einen der folgenden Status haben:

  • Not started
  • Starting
  • Running
  • Draining
  • Paused
  • Failed
  • Failed permanently

In den Logs finden Sie zusätzliche Statusinformationen, z. B. zum Backfill von Tabellen oder zur Anzahl der verarbeiteten Zeilen. Sie können auch die FetchStreamErrors API verwenden, um Fehler abzurufen.

Mit der Discover API verfügbare Objektmetadaten

Die Discover API gibt Objekte zurück, die die Struktur der Objekte darstellen, die in der Datenquelle oder dem Ziel definiert sind, die durch das Verbindungsprofil dargestellt werden. Jedes Objekt hat Metadaten für das Objekt selbst sowie für jedes Datenfeld, das es abruft. Diese Metadaten sind über die Discover API verfügbar.

Nächste Schritte