Change Data Capture da MySQL a BigQuery utilizzando Debezium e Pub/Sub è una pipeline di inserimento flussi che legge i messaggi Pub/Sub con dati di modifica da un database MySQL e scrive i record in BigQuery. Un connettore Debezium acquisisce le modifiche apportate al database MySQL e pubblica i dati modificati in Pub/Sub. Il modello legge quindi i messaggi Pub/Sub e li scrive in BigQuery.
Puoi utilizzare questo modello per sincronizzare i database MySQL e le tabelle BigQuery. La pipeline scrive i dati modificati in una tabella temporanea di BigQuery e aggiorna a intermittenza una tabella BigQuery replicando il database MySQL.
Requisiti della pipeline
- È necessario eseguire il deployment del connettore Debezium.
- I messaggi Pub/Sub devono essere serializzati in una riga Beam Row.
Parametri del modello
Parametri obbligatori
- inputSubscriptions : l'elenco separato da virgole di sottoscrizioni di input Pub/Sub da cui leggere, nel formato
<SUBSCRIPTION_NAME>,<SUBSCRIPTION_NAME>, ...
. - changeLogDataset : il set di dati BigQuery in cui archiviare le tabelle temporanee, nel formato <DATASET_NAME>.
- replicaDataset : la posizione del set di dati BigQuery in cui archiviare le tabelle di replica, nel formato <DATASET_NAME>.
Parametri facoltativi
- inputTopics : elenco separato da virgole di argomenti PubSub a cui viene eseguito il push dei dati CDC.
- updateFrequencySecs : l'intervallo con cui la pipeline aggiorna la tabella BigQuery replicando il database MySQL.
- useSingleTopic : imposta questo valore su true se hai configurato il connettore Debezium per pubblicare tutti gli aggiornamenti della tabella in un singolo argomento. Il valore predefinito è false.
- useStorageWriteApi : se true, la pipeline utilizza l'API BigQuery Storage Scrivi (https://cloud.google.com/bigquery/docs/write-api). Il valore predefinito è
false
. Per ulteriori informazioni, consulta Utilizzo dell'API StorageWrite (https://beam.apache.org/documentation/io/built-in/google-bigquery/#storage-write-api). - useStorageWriteApiAtLeastOnce : quando si utilizza l'API StorageWrite, specifica la semantica della scrittura. Per utilizzare la semantica "at-least-once-semantics" (https://beam.apache.org/documentation/io/built-in/google-bigquery/#at-least-once-semantics), imposta questo parametro su
true
. Per utilizzare la semantica "exactly-once", imposta il parametro sufalse
. Questo parametro si applica solo quandouseStorageWriteApi
ètrue
. Il valore predefinito èfalse
. - numStorageWriteApiStreams : quando si utilizza l'API StorageWrite, specifica il numero di flussi di scrittura. Se
useStorageWriteApi
ètrue
euseStorageWriteApiAtLeastOnce
èfalse
, devi impostare questo parametro. Il valore predefinito è 0. - storageWriteApiTriggeringFrequencySec : quando utilizzi l'API StorageWrite, specifica la frequenza di attivazione in secondi. Se
useStorageWriteApi
ètrue
euseStorageWriteApiAtLeastOnce
èfalse
, devi impostare questo parametro.
Esegui il modello
Per eseguire questo modello, segui questi passaggi:
- Sulla macchina locale, clona il repository DataflowTemplates.
- Passa alla directory
v2/cdc-parent
. - Assicurati che sia stato eseguito il deployment del connettore Debezium.
- Utilizzando Maven, esegui il modello Dataflow:
mvn exec:java -pl cdc-change-applier -Dexec.args="--runner=DataflowRunner \ --inputSubscriptions=SUBSCRIPTIONS \ --updateFrequencySecs=300 \ --changeLogDataset=CHANGELOG_DATASET \ --replicaDataset=REPLICA_DATASET \ --project=PROJECT_ID \ --region=REGION_NAME"
Sostituisci quanto segue:
PROJECT_ID
: l'ID del progetto Google Cloud in cui vuoi eseguire il job DataflowSUBSCRIPTIONS
: elenco separato da virgole di nomi delle sottoscrizioni Pub/SubCHANGELOG_DATASET
: il tuo set di dati BigQuery per i dati del log delle modificheREPLICA_DATASET
: il tuo set di dati BigQuery per le tabelle di replica
Passaggi successivi
- Scopri di più sui modelli Dataflow.
- Consulta l'elenco dei modelli forniti da Google.