Eventi e flussi di dati

La gerarchia dei dati in Datastream è:

  • Uno stream, composto da un'origine dati e da una destinazione.
  • Un oggetto, ovvero una parte di uno stream, ad esempio una tabella di un database specifico.
  • Un evento, ovvero una singola modifica generata da un oggetto specifico, ad esempio un inserimento nel database.

Stream, oggetti ed eventi hanno dati e metadati associati. Questi dati e metadati possono essere utilizzati per diversi scopi.

Informazioni sugli eventi

Ogni evento è costituito da tre tipi di dati:

  • Dati sugli eventi: rappresentano la modifica ai dati stessi dall'oggetto proveniente dall'origine dello stream. Ogni evento contiene l'intera riga modificata.
  • Metadati generici: questi metadati vengono visualizzati in ogni evento generato da Datastream e vengono utilizzati per azioni come la rimozione di dati duplicati nella destinazione.
  • Metadati specifici dell'origine: questi metadati vengono visualizzati in ogni evento generato da un'origine stream specifica. Questi metadati variano in base all'origine.

Dati sull'evento

I dati sugli eventi sono il payload di ogni modifica di un determinato oggetto proveniente da un'origine stream.

Gli eventi sono in formato Avro o JSON.

Quando utilizzi il formato Avro, per ogni colonna l'evento contiene l'indice e il valore della colonna. Utilizzando l'indice della colonna, il nome della colonna e il tipo unificato possono essere recuperati dallo schema nell'intestazione Avro.

Quando utilizzi il formato JSON, per ogni colonna l'evento contiene il nome e il valore della colonna.

I metadati degli eventi possono essere utilizzati per raccogliere informazioni sull'origine dell'evento, nonché per rimuovere i dati duplicati nella destinazione e ordinare gli eventi in base al consumer downstream.

Le tabelle seguenti elencano e descrivono i campi e i tipi di dati per i metadati degli eventi generici e specifici dell'origine.

Metadati generici

Questi metadati sono coerenti in tutti i tipi di stream.

Campo Tipo Avro Tipo JSON Descrizione
stream_name string string Il nome univoco dello stream, come definito al momento della creazione.
read_method string string

Indica se i dati sono stati letti dall'origine utilizzando un metodo Change Data Capture (CDC), nell'ambito di un backfill storico o di un'attività di integrazione creata quando viene eseguito il rollback di una transazione durante la replica CDC.

I valori possibili sono:

  • 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 Il nome utilizzato per raggruppare diversi tipi di eventi, in genere il nome della tabella o dell'oggetto nell'origine.
schema_key string string Un identificatore univoco per lo schema unificato dell'evento.
uuid string string Un identificatore univoco dell'evento generato da Datastream.
read_timestamp timestamp-millis string Il timestamp (UTC) in cui il record è stato letto da Datastream (il timestamp dell'epoca in millisecondi).
source_timestamp timestamp-millis string Il timestamp (UTC) della modifica del record nell'origine (il timestamp epoch in millisecondi).
sort_keys {"type": "array", "items": ["string", "long"]} matrice Un array di valori che può essere utilizzato per ordinare gli eventi nell'ordine in cui si sono verificati.

Metadati specifici dell'origine

Questi metadati sono associati agli eventi CDC e di backfill di un database di origine. Per visualizzare questi metadati, seleziona un'origine dal menu a discesa di seguito.

Origine Campo Tipo Avro Tipo JSON Descrizione
MySQL log_file string string Il file di log da cui Datastream estrae gli eventi nella replica CDC quando utilizza la replica basata su binlog.
MySQL log_position Lungo Lungo La posizione del log (offset) nel log binario MySQL quando si utilizza la replica basata su binlog.
MySQL binlog_file string string Il file di log binario da cui Datastream estrae gli eventi nella replica CDC quando utilizza la replica basata su GTID.
MySQL binlog_position Lungo Lungo La posizione (offset) nel log binario MySQL quando si utilizza la replica basata su GTID.
MySQL gtid string string L'ID transazione globale (GTID) che identifica in modo univoco la transazione per l'evento quando si utilizza la replica basata su GTID.
MySQL datastream_master_server_uuid string string L'UUID dell'istanza del server di origine MySQL da cui Datastream legge gli eventi quando utilizza la replica basata su GTID.
MySQL datastream_master_server_id Lungo Lungo L'ID server dell'istanza del server di origine MySQL da cui Datastream legge gli eventi quando utilizza la replica basata su GTID.
MySQL primary_keys array di stringhe array di stringhe L'elenco di uno o più nomi di colonne che compongono la chiave primaria della tabella. Se la tabella non ha una chiave primaria, questo campo è vuoto.
MySQL is_deleted boolean boolean
  • Un valore true indica che la riga è stata eliminata nell'origine.
  • Un valore false indica che la riga non è stata eliminata.
MySQL database string string Il database associato all'evento.
MySQL table string string La tabella associata all'evento.
MySQL change_type string string

Il tipo di modifica (INSERT, UPDATE-INSERT, UPDATE-DELETE e DELETE) rappresentato dall'evento.
Oracle log_file string string Il file di log da cui Datastream estrae gli eventi nella replica CDC.
Oracle scn Lungo Lungo La posizione del log (offset) nel log delle transazioni Oracle.
Oracle row_id string string row_id di Oracle.
Oracle is_deleted boolean boolean
  • Un valore true indica che la riga è stata eliminata nell'origine.
  • Un valore false indica che la riga non è stata eliminata.
Oracle database string string Il database associato all'evento.
Oracle schema string string Lo schema associato alla tabella dell'evento.
Oracle table string string La tabella associata all'evento.
Oracle change_type string string

Il tipo di modifica (INSERT, UPDATE-INSERT, UPDATE-DELETE e DELETE) rappresentato dall'evento.
Oracle tx_id string string L'ID transazione a cui appartiene l'evento.
Oracle rs_id string string L'ID del set di record. L'accoppiamento di rs_id e ssn identifica in modo univoco una riga in V$LOGMNR_CONTENTS. rs_id identifica in modo univoco il record di ripetizione che ha generato la riga.
Oracle ssn Lungo Lungo Un numero di sequenza SQL. Questo numero viene utilizzato con rs_id e identifica in modo univoco una riga in V$LOGMNR_CONTENTS.
PostgreSQL schema string string Lo schema associato alla tabella dell'evento.
PostgreSQL table string string La tabella associata all'evento.
PostgreSQL is_deleted boolean boolean
  • Un valore true indica che la riga è stata eliminata nell'origine.
  • Un valore false indica che la riga non è stata eliminata.
PostgreSQL change_type string string Il tipo di modifica (INSERT, UPDATE, DELETE) rappresentato dall'evento.
PostgreSQL tx_id string string L'ID transazione a cui appartiene l'evento.
PostgreSQL lsn string string Il numero di sequenza di log per la voce corrente.
PostgreSQL primary_keys array di stringhe array di stringhe L'elenco di uno o più nomi di colonne che compongono la chiave primaria della tabella. Se la tabella non ha una chiave primaria, questo campo è vuoto.
SQL Server table string string La tabella associata all'evento.
SQL Server database Lungo Lungo Il database associato all'evento.
SQL Server schema array di stringhe array di stringhe Lo schema associato alla tabella dell'evento.
SQL Server is_deleted boolean boolean
  • Un valore true indica che la riga è stata eliminata nell'origine.
  • Un valore false indica che la riga non è stata eliminata.
SQL Server lsn string string Il numero di sequenza del log per l'evento.
SQL Server tx_id string string L'ID transazione a cui appartiene l'evento.
SQL Server physical_location array di numeri interi array di numeri interi La posizione fisica del record di log descritta da tre numeri interi: ID file, ID pagina e ID slot del record.
SQL Server replication_index array di stringhe array di stringhe L'elenco dei nomi delle colonne di un indice che possono identificare in modo univoco una riga nella tabella.
SQL Server change_type string string

Il tipo di modifica (INSERT, UPDATE, DELETE) rappresentato dall'evento.
Spanner commit_timestamp timestamp-micros string Il timestamp di commit dell'evento.
Spanner snapshot boolean boolean Indica se l'evento è un evento di snapshot di backfill.
Spanner project_id string string L'identificatore del progetto Spanner.
Spanner instance_id string string L'identificatore dell'istanza Spanner.
Spanner database_id string string L'identificatore del database Spanner.
Spanner schema string string Lo schema associato alla tabella dell'evento. Vuoto se la tabella non si trova in uno schema denominato.
Spanner table string string La tabella Spanner.
Spanner change_stream_name string string Il flusso di modifiche in tempo reale Spanner.
Spanner server_transaction_id string string L'identificatore della transazione associato all'evento.
Spanner record_sequence long o string long o string La sequenza di record associata all'evento. Per le modifiche in tempo reale create con la modalità di partizionamento IMMUTABLE_KEY_RANGE, il valore è un numero di sequenza numerico di tipo long. Per i flussi di modifiche creati con la modalità di partizione MUTABLE_KEY_RANGE, il valore è la stringa record_sequence del record di modifica dei dati, ad esempio 1cac8f3f38eeaf53-00000000.
Spanner mod_number Lungo Lungo Il numero di modifica associato all'evento.
Spanner transaction_tag string string Il tag della transazione associata.
Spanner number_of_records_in_transaction Lungo Lungo Il numero di record nella transazione associata.
Spanner value_capture_type string string Il tipo di acquisizione del valore della modifica in tempo reale. I valori possibili sono NEW_ROW e NEW_ROW_AND_OLD_VALUES.
Spanner change_type string string Il tipo di modifica (INSERT, UPDATE o DELETE) rappresentato dall'evento.
Spanner primary_keys array di stringhe array di stringhe L'elenco delle colonne di chiave primaria della tabella.
Spanner is_deleted boolean boolean Se la modifica rappresenta un evento DELETE.
Salesforce object_name string string

Il nome dell'oggetto Salesforce associato all'evento.
Salesforce domain string string

Il nome del dominio associato all'evento.
Salesforce is_deleted boolean boolean
  • Un valore true indica che la riga è stata eliminata nell'origine.
  • Un valore false indica che la riga non è stata eliminata.
Salesforce change_type string string

Il tipo di modifica (INSERT, UPDATE, DELETE) rappresentato dall'evento.
Salesforce primary_keys array di stringhe array di stringhe L'elenco dei nomi delle colonne che compongono la chiave primaria delle tabelle. Se la tabella non ha una chiave primaria, questo campo è vuoto.
MongoDB database string string Il database associato all'evento.
MongoDB collection string string La raccolta associata all'evento. Le raccolte sono analoghe alle tabelle nei database relazionali.
MongoDB change_type string string Il tipo di modifica (CREATE, UPDATE e DELETE) rappresentato dall'evento.
MongoDB is_deleted boolean boolean
  • Un valore true indica che la riga è stata eliminata nell'origine.
  • Un valore false indica che la riga non è stata eliminata.
MongoDB primary_keys array di stringhe array di stringhe Il campo _id, che funge da chiave primaria per ogni documento di una raccolta.

Esempio di flusso di eventi

Questo flusso illustra gli eventi generati da tre operazioni consecutive: INSERT, UPDATE e DELETE, su una singola riga di una tabella SAMPLE per un database di origine.

TEMPO THIS_IS_MY_PK (int) FIELD1 (nchar nullable) FIELD2 (nchar non-null)>
0 1231535353 foo TLV
1 1231535353 NULL TLV

INSERT (T0)

Il payload del messaggio è costituito dall'intera nuova riga.

{
  "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)

Il payload del messaggio è costituito dall'intera nuova riga. Non include i valori precedenti.

{
  "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)

Il payload del messaggio è costituito dall'intera nuova riga.

{
  "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",
  }
}

Ordinamento e coerenza

Questa sezione spiega come Datastream gestisce l'ordinamento e la coerenza.

Ordine

Datastream non garantisce l'invio degli eventi in ordine, ma ogni evento contiene la riga completa dei dati, il timestamp di quando i dati sono stati scritti nell'origine e i metadati di ordinamento.

Gli eventi che fanno parte di un backfill iniziale hanno il campo read_method impostato su backfill o che termina con -backfill, -fulldump o -incremental. Gli eventi di backfill vengono generati in ordine sparso per impostazione predefinita e possono essere utilizzati in qualsiasi ordine.

Gli eventi che fanno parte della replica di Change Data Capture (CDC) in corso hanno il campo read_method impostato su cdc o su un valore CDC specifico dell'origine, ad esempio mysql-cdc-binlog, oracle-cdc-logminer, postgres-cdc-wal, sqlserver-cdc o salesforce-cdc. Il modo in cui ordini gli eventi di replica in corso in tutte le origini dipende dalla destinazione:

Destinazione Descrizione
BigQuery

L'ordine degli eventi in BigQuery dipende dalla modalità di scrittura configurata per lo stream:

  • Modalità di scrittura Unisci: BigQuery unisce automaticamente gli eventi non in ordine nella sequenza corretta utilizzando i metadati degli eventi e un numero di sequenza di modifica (CSN) interno. Le tabelle senza una chiave primaria vengono scritte in modalità di sola aggiunta.
  • Modalità di scrittura di sola aggiunta, incluse le tabelle Apache Iceberg: Datastream aggiunge gli eventi di modifica alla tabella di destinazione e compila la colonna datastream_metadata STRUCT. Per determinare l'ordine in cui si sono verificati gli eventi nell'origine, ordinali utilizzando il campo datastream_metadata.change_sequence_number o il campo dell'array datastream_metadata.sort_keys.
Cloud Storage

In Cloud Storage, gli eventi dello stesso periodo di tempo possono estendersi su più file. Per determinare l'ordine in cui si sono verificati gli eventi nell'origine, ordinali utilizzando il campo dell'array sort_keys nei metadati generici.

Coerenza

Datastream garantisce che i dati del database di origine vengano inviati alla destinazione almeno una volta. Nessun evento viene perso, ma esiste la possibilità di eventi duplicati nello stream. La finestra per gli eventi duplicati deve essere dell'ordine di minuti e l'identificatore univoco universale (UUID) dell'evento nei metadati dell'evento può essere utilizzato per rilevare i duplicati.

Tuttavia, non fare affidamento esclusivamente sull'UUID per deduplicare gli eventi a valle. Se si verifica un'interruzione della pipeline, Datastream rilegge il log di origine dall'ultimo checkpoint salvato per garantire la distribuzione. Poiché Datastream genera dinamicamente gli UUID quando legge un evento, questo tipo di nuovo tentativo genera nuovi UUID per gli stessi dati di origine. Inoltre, durante le operazioni collettive nel database di origine, più righe distinte potrebbero condividere lo stesso numero di sequenza di modifica (CSN).

Per deduplicare i dati in modo affidabile, partiziona gli eventi in base alle chiavi primarie della tabella di origine e raggruppali utilizzando il numero di sequenza di modifica (CSN) (il campo change_sequence_number). In questo modo, i nuovi tentativi dello stesso evento vengono raggruppati correttamente, mentre gli aggiornamenti delle righe legittimamente distinti delle transazioni collettive vengono conservati.

Quando i file di log del database contengono transazioni non sottoposte a commit, se viene eseguito il rollback di una transazione, il database lo riflette nei file di log come operazioni di data manipulation language (DML) "inverse". Ad esempio, un'operazione INSERT di cui è stato eseguito il rollback avrà un'operazione DELETE corrispondente. Datastream legge queste operazioni dai file di log.

Informazioni sugli stream

Ogni stream ha metadati che descrivono sia lo stream sia l'origine da cui estrae i dati. Questi metadati includono informazioni come il nome dello stream e i profili di connessione di origine e destinazione.

Per visualizzare la definizione completa dell'oggetto Stream, consulta la documentazione di riferimento dell'API.

Stato e condizione dello stream

Uno stream può avere uno dei seguenti stati:

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

Puoi utilizzare i log per trovare ulteriori informazioni sullo stato, ad esempio il backfill delle tabelle o il numero di righe elaborate. Puoi anche utilizzare l'API FetchStreamErrors per recuperare gli errori.

Metadati degli oggetti disponibili tramite l'API Discover

L'API di rilevamento restituisce oggetti che rappresentano la struttura degli oggetti definiti nell'origine o nella destinazione dei dati rappresentata dal profilo di connessione. Ogni oggetto ha metadati sull'oggetto stesso, nonché per ogni campo di dati che estrae. Questi metadati sono disponibili tramite l'API Discover.

Passaggi successivi