Modifica l'acquisizione dei dati da MySQL a BigQuery utilizzando il modello Debezium e Pub/Sub (stream)

Il modello Change Data Capture (CDC) da MySQL a BigQuery utilizzando il modello Debezium e Pub/Sub è una pipeline in modalità flusso che legge i messaggi Pub/Sub con i dati delle modifiche da un database MySQL e scrive i record in BigQuery. Un connettore Debezium acquisisce le modifiche 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 BigQuery e aggiorna a intermittenza una tabella BigQuery che replica il database MySQL.

Requisiti della pipeline

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 Pub/Sub 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 delle tabelle in un singolo argomento. Il valore predefinito è: false.
  • useStorageWriteApi : se il valore è true, la pipeline utilizza l'API BigQuery Storage Write (https://cloud.google.com/bigquery/docs/write-api). Il valore predefinito è false. Per saperne di più, consulta la pagina relativa all'utilizzo dell'API Storage Write (https://beam.apache.org/documentation/io/built-in/google-bigquery/#storage-write-api).
  • useStorageWriteApiAtLeastOnce : quando utilizzi l'API Storage Write, specifica la semantica della scrittura. Per utilizzare almeno la semantica (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 su false. Questo parametro si applica solo quando useStorageWriteApi è true. Il valore predefinito è false.
  • numStorageWriteApiStreams : quando si utilizza l'API Storage Write, specifica il numero di flussi di scrittura. Se useStorageWriteApi è true e useStorageWriteApiAtLeastOnce è false, devi impostare questo parametro. Il valore predefinito è 0.
  • storageWriteApiTriggeringFrequencySec : quando utilizzi l'API Storage Write, specifica la frequenza di attivazione in secondi. Se useStorageWriteApi è true e useStorageWriteApiAtLeastOnce è false, devi impostare questo parametro.

Esegui il modello

Per eseguire questo modello, segui questi passaggi:

  1. Sulla macchina locale, clona il repository DataflowTemplates.
  2. Passa alla directory v2/cdc-parent.
  3. Assicurati che sia stato eseguito il deployment del connettore Debezium.
  4. 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 Dataflow
    • SUBSCRIPTIONS: il tuo elenco separato da virgole di nomi di sottoscrizione Pub/Sub
    • CHANGELOG_DATASET: il tuo set di dati BigQuery per i dati del log delle modifiche
    • REPLICA_DATASET: il tuo set di dati BigQuery per le tabelle di replica

Passaggi successivi