Die Datenhierarchie in Datastream besteht aus folgenden Elementen:
- Ein Stream, der aus einer Datenquelle und einem Ziel besteht.
- Objekt: ein Teil eines Streams, z. B. eine Tabelle aus einer bestimmten Datenbank.
- Ereignis: eine einzelne Änderung, die von einem bestimmten Objekt generiert wird, z. B. eine Datenbankeinfügung.
Streams, Objekte und Ereignisse sind mit Daten und Metadaten verknüpft. Diese Daten und Metadaten können für verschiedene Zwecke verwendet werden.
Ü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:
|
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 |
|
| 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 ( |
| 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 |
|
| 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 ( |
| 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 |
|
| 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 |
|
| 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 ( |
| Spanner | commit_timestamp |
String | 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 | change_stream_name |
String | String | Der Cloud Spanner-Änderungsstream. |
| Spanner | table |
String | String | Die Cloud Spanner-Tabelle. |
| Spanner | server_transaction_id |
String | String | Die Transaktions-ID, die mit dem Ereignis verknüpft ist. |
| Spanner | record_sequence |
String | String | Die mit dem Ereignis verknüpfte Datensatzsequenz. |
| Spanner | mod_index |
String | String | Die Änderungsnummer, die dem Ereignis zugeordnet ist. |
| Spanner | transaction_tag |
String | String | Das Tag für die zugehörige Transaktion. |
| Spanner | system_transaction |
String | String | Gibt an, ob die Transaktion eine Systemtransaktion ist. |
| Spanner | number_of_records_in_transaction |
long | long | Die Anzahl der Datensätze in der zugehörigen Transaktion. |
| Spanner | value_capture_type |
enum | enum | Der Werterfassungstyp des Änderungsstreams. |
| Spanner | mod_type |
enum | enum | Gibt an, ob der Eintrag ein INSERT, UPDATE oder DELETE war. |
| Spanner | primary_keys |
Array | 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 |
|
| Salesforce | change_type |
String | String | Die Art der Änderung ( |
| 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 |
|
| 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 erläutert, wie Datastream die Sortierung und Konsistenz verarbeitet.
Bestellung
Datastream garantiert keine Reihenfolge, aber jedes Ereignis enthält die vollständige Datenzeile und den Zeitstempel, zu dem die Daten in die Quelle geschrieben wurden. In BigQuery werden Ereignisse, die nicht in der richtigen Reihenfolge sind, automatisch in der richtigen Reihenfolge zusammengeführt. BigQuery verwendet die Ereignismetadaten und eine interne Änderungssequenznummer (CSN), um die Ereignisse in der richtigen Reihenfolge auf die Tabelle anzuwenden. In Cloud Storage können Ereignisse aus demselben Zeitraum mehrere Dateien umfassen.
In der falschen Reihenfolge generierte Ereignisse erfolgen standardmäßig, wenn ein Backfill für den ersten Backfill der Daten durchgeführt wird, die beim Initiieren des Streams erstellt wurden.
Die Reihenfolge kann für jede Quelle einzeln abgeleitet werden.
| Quelle | Beschreibung |
|---|---|
| MySQL | Bei Ereignissen, die Teil des ersten Backfills sind, beginnt das Feld Bei Ereignissen, die Teil einer laufenden Replikation sind, ist das Feld Bei der binlog-basierten Replikation kann die Reihenfolge aus der Kombination der Felder |
| Oracle | Ereignisse, die Teil des ersten Backfills sind, haben das Feld Bei Ereignissen, die Teil der laufenden Replikation sind, ist das Feld Die Reihenfolge kann aus der Kombination des Felds |
| PostgreSQL | Ereignisse, die Teil des ersten Backfills sind, haben das Feld Bei Ereignissen, die Teil der laufenden Replikation sind, ist das Feld Die Reihenfolge kann aus der Kombination des Felds |
| SQL Server |
Ereignisse, die Teil des ersten Backfills sind, haben das Feld Bei Ereignissen, die Teil der laufenden Replikation sind, ist das Feld Die Reihenfolge kann aus der Kombination des Felds Bei der Replikation in BigQuery wird die Erstellungszeit der Datenbank auch verwendet, um die richtige Reihenfolge sicherzustellen. Das ist wichtig in Szenarien, in denen sich die Quelldatenbankinstanz ändern kann, z. B. bei Migrationen oder Failover zu einem Replikat. |
| Salesforce |
Die Reihenfolge kann mithilfe des |
| MongoDB |
Die Reihenfolge kann mithilfe des Felds |
| Spanner |
Die Reihenfolge kann mithilfe der Kombination der Felder |
Konsistenz
Datastream sorgt dafür, dass die Daten aus der Quelldatenbank mindestens einmal an das Ziel gesendet 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 startedStartingRunningDrainingPausedFailedFailed 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
- Weitere Informationen zu Streams finden Sie unter Stream-Lebenszyklus.
- Informationen zum Erstellen eines Streams finden Sie unter Stream erstellen.