Modifiche in tempo reale di Bigtable nel modello Pub/Sub

Il modello di modifiche in tempo reale di Bigtable in Pub/Sub è una pipeline di inserimento flussi che trasmette il flusso dei record delle modifiche dei dati di Bigtable e li pubblica in un argomento Pub/Sub utilizzando Dataflow.

Una modifica in tempo reale di Bigtable ti consente di sottoscrivere le mutazioni dei dati in base alla tabella. Quando ti abboni ai modifiche in tempo reale delle tabelle, si applicano i seguenti vincoli:

  • Vengono restituiti solo le celle modificate e i descrittori delle operazioni di eliminazione.
  • Viene restituito solo il nuovo valore di una cella modificata.

Quando i record delle modifiche dei dati vengono pubblicati in un argomento Pub/Sub, i messaggi potrebbero essere inseriti nell'ordine sbagliato rispetto all'ordine del timestamp di commit originale di Bigtable.

I record delle modifiche dei dati di Bigtable che non possono essere pubblicati negli argomenti Pub/Sub vengono temporaneamente inseriti in una directory di code di messaggi non recapitabili (coda dei messaggi non elaborati) in Cloud Storage. Dopo il numero massimo di nuovi tentativi non riusciti, questi record vengono inseriti a tempo indeterminato nella stessa directory della coda dei messaggi non recapitabili per la revisione da parte di persone fisiche o per un'ulteriore elaborazione da parte dell'utente.

La pipeline richiede l'esistenza dell'argomento Pub/Sub di destinazione. L'argomento di destinazione potrebbe essere configurato per convalidare i messaggi utilizzando uno schema. Quando un argomento Pub/Sub specifica uno schema, la pipeline si avvia solo se lo schema è valido. A seconda del tipo di schema, utilizza una delle seguenti definizioni dello schema per l'argomento di destinazione:

buffer di protocollo

syntax = "proto2";

package com.google.cloud.teleport.bigtable;

option java_outer_classname = "ChangeLogEntryProto";

message ChangelogEntryProto{
  required bytes rowKey = 1;
  enum ModType {
    SET_CELL = 0;
    DELETE_FAMILY = 1;
    DELETE_CELLS = 2;
    UNKNOWN = 3;
  }
  required ModType modType = 2;
  required bool isGC = 3;
  required int32 tieBreaker = 4;
  required int64 commitTimestamp = 5;
  required string columnFamily = 6;
  optional bytes column = 7;
  optional int64 timestamp = 8;
  optional int64 timestampFrom = 9;
  optional int64 timestampTo = 10;
  optional bytes value = 11;
  required string sourceInstance = 12;
  required string sourceCluster = 13;
  required string sourceTable = 14;
}

Avro

{
    "name" : "ChangelogEntryMessage",
    "type" : "record",
    "namespace" : "com.google.cloud.teleport.bigtable",
    "fields" : [
      { "name" : "rowKey", "type" : "bytes"},
      {
        "name" : "modType",
        "type" : {
          "name": "ModType",
          "type": "enum",
          "symbols": ["SET_CELL", "DELETE_FAMILY", "DELETE_CELLS", "UNKNOWN"]}
      },
      { "name": "isGC", "type": "boolean" },
      { "name": "tieBreaker", "type": "int"},
      { "name": "columnFamily", "type": "string"},
      { "name": "commitTimestamp", "type" : "long"},
      { "name" : "sourceInstance", "type" : "string"},
      { "name" : "sourceCluster", "type" : "string"},
      { "name" : "sourceTable", "type" : "string"},
      { "name": "column", "type" : ["null", "bytes"]},
      { "name": "timestamp", "type" : ["null", "long"]},
      { "name": "timestampFrom", "type" : ["null", "long"]},
      { "name": "timestampTo", "type" : ["null", "long"]},
      { "name" : "value", "type" : ["null", "bytes"]}
   ]
}

JSON

Utilizza il seguente schema Protobuf con codifica dei messaggi JSON:

syntax = "proto2";

package com.google.cloud.teleport.bigtable;

option java_outer_classname = "ChangelogEntryMessageText";

message ChangelogEntryText{
  required string rowKey = 1;
  enum ModType {
    SET_CELL = 0;
    DELETE_FAMILY = 1;
    DELETE_CELLS = 2;
    UNKNOWN = 3;
  }
  required ModType modType = 2;
  required bool isGC = 3;
  required int32 tieBreaker = 4;
  required int64 commitTimestamp = 5;
  required string columnFamily = 6;
  optional string column = 7;
  optional int64 timestamp = 8;
  optional int64 timestampFrom = 9;
  optional int64 timestampTo = 10;
  optional string value = 11;
  required string sourceInstance = 12;
  required string sourceCluster = 13;
  required string sourceTable = 14;
}

Ogni nuovo messaggio Pub/Sub include una voce di un record di modifica dei dati restituito dal flusso di modifiche dalla riga corrispondente nella tabella Bigtable. Il modello Pub/Sub suddivide le voci in ogni record delle modifiche dei dati in singole modifiche a livello di cella.

Descrizione del messaggio di output Pub/Sub

Nome campo Descrizione
rowKey La chiave di riga della riga modificata. Arriva sotto forma di array di byte. Quando è configurata la codifica dei messaggi JSON, le chiavi di riga vengono restituite come stringhe. Se useBase64Rowkeys è specificato, le chiavi di riga sono codificate in Base64. Altrimenti, un set di caratteri specificato da bigtableChangeStreamCharset viene utilizzato per decodificare i byte della chiave di riga in una stringa.
modType Il tipo di mutazione della riga. Utilizza uno dei seguenti valori: SET_CELL, DELETE_CELLS o DELETE_FAMILY.
columnFamily La famiglia di colonne interessata dalla modifica della riga.
column Il qualificatore di colonna interessato dalla modifica della riga. Per il tipo di mutazione DELETE_FAMILY, il campo della colonna non è impostato. Arriva sotto forma di array di byte. Quando è configurata la codifica dei messaggi JSON, le colonne vengono restituite come stringhe. Se useBase64ColumnQualifier è specificato, il campo colonna ha una codifica Base64. Altrimenti, un set di caratteri specificato da bigtableChangeStreamCharset viene utilizzato per decodificare i byte della chiave di riga in una stringa.
commitTimestamp L'ora in cui Bigtable applica la mutazione. Il tempo è misurato in microsecondi dall'epoca di Unix (1 gennaio 1970 alle ore UTC).
timestamp Il valore del timestamp della cella interessata dalla mutazione. Per i tipi di mutazione DELETE_CELLS e DELETE_FAMILY, il timestamp non è impostato. Il tempo è misurato in microsecondi dall'epoca di Unix (1 gennaio 1970 alle ore UTC).
timestampFrom Descrive un inizio inclusivo dell'intervallo di timestamp per tutte le celle eliminate dalla mutazione DELETE_CELLS. Per altri tipi di mutazioni, timestampFrom non è impostato. Il tempo è misurato in microsecondi dall'epoca di Unix (1 gennaio 1970 alle ore UTC).
timestampTo Descrive una fine esclusiva dell'intervallo del timestamp per tutte le celle eliminate dalla mutazione DELETE_CELLS. Per altri tipi di mutazioni, timestampTo non è impostato.
isGC Un valore booleano che indica se la mutazione è generata da un meccanismo di garbage collection Bigtable.
tieBreaker Quando due mutazioni vengono registrate contemporaneamente da cluster Bigtable diversi, alla tabella di origine viene applicata la mutazione con il valore tiebreaker più alto. Le mutazioni con valori tiebreaker più bassi vengono ignorate.
value Il nuovo valore impostato dalla mutazione. A meno che non sia impostata l'opzione della pipeline stripValues, il valore viene impostato per le modifiche SET_CELL. Per altri tipi di mutazione, il valore non è impostato. Arriva sotto forma di array di byte. Quando è configurata la codifica dei messaggi JSON, i valori vengono restituiti come stringhe. Quando si specifica useBase64Values, il valore è codificato in Base64. Altrimenti, un set di caratteri specificato da bigtableChangeStreamCharset viene utilizzato per decodificare i byte dei valori in una stringa.
sourceInstance Il nome dell'istanza Bigtable che ha registrato la mutazione. Potrebbe verificarsi quando più pipeline trasmettono modifiche da istanze diverse allo stesso argomento Pub/Sub.
sourceCluster Il nome del cluster Bigtable che ha registrato la mutazione. Può essere utilizzato quando più pipeline trasmettono modifiche da istanze diverse allo stesso argomento Pub/Sub.
sourceTable Il nome della tabella Bigtable che ha ricevuto la mutazione. Può essere utilizzato nel caso in cui più pipeline trasmettono le modifiche da tabelle diverse allo stesso argomento Pub/Sub.

Requisiti della pipeline

  • L'istanza di origine Bigtable specificata.
  • La tabella di origine Bigtable specificata. Nella tabella devono essere abilitati modifiche in tempo reale.
  • Il profilo dell'applicazione Bigtable specificato.
  • Deve esistere l'argomento Pub/Sub specificato.

Parametri del modello

Parametri obbligatori

  • pubSubTopic : il nome dell'argomento Pub/Sub di destinazione.
  • bigtableChangeStreamAppProfile : l'ID profilo dell'applicazione Bigtable. Il profilo di applicazione deve utilizzare il routing a cluster singolo e consentire transazioni su riga singola.
  • bigtableReadInstanceId : ID dell'istanza Bigtable di origine.
  • bigtableReadTableId : l'ID della tabella Bigtable di origine.

Parametri facoltativi

  • messageEncoding : la codifica dei messaggi da pubblicare nell'argomento Pub/Sub. Quando è configurato lo schema dell'argomento di destinazione, la codifica dei messaggi viene determinata dalle impostazioni dell'argomento. Sono supportati i seguenti valori: BINARY e JSON. Il valore predefinito è JSON.
  • messageFormat : la codifica dei messaggi da pubblicare nell'argomento Pub/Sub. Quando è configurato lo schema dell'argomento di destinazione, la codifica dei messaggi viene determinata dalle impostazioni dell'argomento. Sono supportati i seguenti valori: AVRO, PROTOCOL_BUFFERS e JSON. Il valore predefinito è JSON. Quando viene utilizzato il formato JSON, i campi rowKey, colonna e valore del messaggio sono stringhe il cui contenuto è determinato dalle opzioni useBase64Rowkeys, useBase64ColumnQualifiers, useBase64Values e bigtableChangeStreamCharset della pipeline.
  • stripValues : se questo criterio viene impostato su true, le mutazioni di SET_cell vengono restituite senza che siano stati impostati nuovi valori. Il valore predefinito è false. Questo parametro è utile quando non è necessario che sia presente un nuovo valore, noto anche come invalidazione della cache, o quando i valori sono molto grandi e superano i limiti di dimensione dei messaggi Pub/Sub.
  • dlqDirectory : la directory per la coda dei messaggi non recapitabili. I record che non vengono elaborati vengono archiviati in questa directory. Il valore predefinito è una directory nella località temporanea del job Dataflow. Nella maggior parte dei casi, puoi utilizzare il percorso predefinito.
  • dlqRetryMinutes : il numero di minuti tra i nuovi tentativi in coda dei messaggi non recapitabili. Il valore predefinito è 10.
  • dlqMaxRetries : il numero massimo di nuovi tentativi per i messaggi non recapitabili. Il valore predefinito è 5.
  • useBase64Rowkeys : utilizzato con la codifica dei messaggi JSON. Se impostato su true, il campo rowKey è una stringa con codifica Base64. Altrimenti, il valore rowKey viene generato utilizzando bigtableChangeStreamCharset per decodificare i byte in una stringa. Il valore predefinito è false.
  • pubSubProjectId : l'ID progetto Bigtable. Il valore predefinito è il progetto del job Dataflow.
  • useBase64ColumnQualifiers : utilizzato con la codifica dei messaggi JSON. Se impostato su true, il campo column è una stringa con codifica Base64. Altrimenti, la colonna viene prodotta utilizzando bigtableChangeStreamCharset per decodificare i byte in una stringa. Il valore predefinito è false.
  • useBase64Values : utilizzato con la codifica dei messaggi JSON. Se impostato su true, il campo del valore è una stringa con codifica Base64. In caso contrario, il valore viene prodotto utilizzando bigtableChangeStreamCharset per decodificare i byte in una stringa. Il valore predefinito è false.
  • disableDlqRetries : indica se disabilitare o meno i nuovi tentativi per il DLQ. Il valore predefinito è: false.
  • bigtableChangeStreamMetadataInstanceId : la modifica di Bigtable trasmette l'ID istanza dei metadati. Il valore predefinito è vuoto.
  • bigtableChangeStreamMetadataTableTableId : l'ID della tabella dei metadati del connettore dei modifiche in tempo reale Bigtable. Se non viene fornita, una tabella dei metadati del connettore di modifiche in tempo reale di Bigtable viene creata automaticamente durante l'esecuzione della pipeline. Il valore predefinito è vuoto.
  • bigtableChangeStreamCharset : il nome del set di caratteri modifiche in tempo reale di Bigtable. Il valore predefinito è UTF-8.
  • bigtableChangeStreamStartTimestamp : il timestamp iniziale (https://tools.ietf.org/html/rfc3339), inclusivo, da utilizzare per leggere le modifiche in tempo reale. Ad esempio: 2022-05-05T07:59:59Z. Il valore predefinito è il timestamp dell'ora di inizio della pipeline.
  • bigtableChangeStreamIgnoreColumnFamilies : un elenco separato da virgole di nomi di famiglia di colonne da ignorare. Il valore predefinito è vuoto.
  • bigtableChangeStreamIgnoreColumns : un elenco separato da virgole di modifiche dei nomi delle colonne da ignorare. Il valore predefinito è vuoto.
  • bigtableChangeStreamName : un nome univoco per la pipeline client. Consente di riprendere l'elaborazione dal punto in cui una pipeline precedentemente in esecuzione è stata arrestata. Il valore predefinito è un nome generato automaticamente. Consulta i log dei job Dataflow per il valore utilizzato.
  • bigtableChangeStreamResume : se l'impostazione è impostata su true, l'elaborazione di una nuova pipeline riprende dal punto in cui è stata arrestata una pipeline in esecuzione in precedenza con lo stesso valore bigtableChangeStreamName. Se la pipeline con il valore bigtableChangeStreamName specificato non è mai stata eseguita, non viene avviata una nuova pipeline. Se la norma è impostata su false, viene avviata una nuova pipeline. Se una pipeline con lo stesso valore bigtableChangeStreamName è già stata eseguita per l'origine specificata, non viene avviata una nuova pipeline. Il valore predefinito è false.
  • bigtableReadProjectId : l'ID progetto Bigtable. Il valore predefinito è il progetto per il job Dataflow.

Esegui il modello

Console

  1. Vai alla pagina Crea job da modello di Dataflow.
    2. Vai a Crea job da modello
  2. Nel campo Nome job, inserisci un nome univoco per il job.
  3. (Facoltativo) Per Endpoint a livello di regione, seleziona un valore dal menu a discesa. La regione predefinita è us-central1.

    Per un elenco delle regioni in cui puoi eseguire un job Dataflow, vedi Località Dataflow.

  4. Nel menu a discesa Modello Dataflow, seleziona the Bigtable change streams to Pub/Sub template.
  5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
  6. Fai clic su Esegui job.

gcloud

Nella shell o nel terminale, esegui il modello:

gcloud dataflow flex-template run JOB_NAME \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Bigtable_Change_Streams_to_PubSub \
    --parameters \
bigtableReadInstanceId=BIGTABLE_INSTANCE_ID,\
bigtableReadTableId=BIGTABLE_TABLE_ID,\
bigtableChangeStreamAppProfile=BIGTABLE_APPLICATION_PROFILE_ID,\
pubSubTopic=PUBSUB_TOPIC

Sostituisci quanto segue:

  • PROJECT_ID: l'ID del progetto Google Cloud in cui vuoi eseguire il job Dataflow
  • JOB_NAME: un nome job univoco a tua scelta
  • VERSION: la versione del modello che vuoi utilizzare

    Puoi utilizzare i seguenti valori:

  • REGION_NAME: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
  • BIGTABLE_INSTANCE_ID: ID istanza Bigtable.
  • BIGTABLE_TABLE_ID: l'ID della tua tabella Bigtable.
  • BIGTABLE_APPLICATION_PROFILE_ID: ID del profilo dell'applicazione Bigtable.
  • PUBSUB_TOPIC: nome dell'argomento di destinazione Pub/Sub

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per maggiori informazioni sull'API e sui relativi ambiti di autorizzazione, consulta projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
  "launch_parameter": {
    "jobName": "JOB_NAME",
    "containerSpecGcsPath": "gs://dataflow-templates-REGION_NAME/VERSION/flex/Bigtable_Change_Streams_to_PubSub",
    "parameters": {
        "bigtableReadInstanceId": "BIGTABLE_INSTANCE_ID",
        "bigtableReadTableId": "BIGTABLE_TABLE_ID",
        "bigtableChangeStreamAppProfile": "BIGTABLE_APPLICATION_PROFILE_ID",
        "pubSubTopic": "PUBSUB_TOPIC"
    }
  }
}

Sostituisci quanto segue:

  • PROJECT_ID: l'ID del progetto Google Cloud in cui vuoi eseguire il job Dataflow
  • JOB_NAME: un nome job univoco a tua scelta
  • VERSION: la versione del modello che vuoi utilizzare

    Puoi utilizzare i seguenti valori:

  • LOCATION: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
  • BIGTABLE_INSTANCE_ID: ID istanza Bigtable.
  • BIGTABLE_TABLE_ID: l'ID della tua tabella Bigtable.
  • BIGTABLE_APPLICATION_PROFILE_ID: ID del profilo dell'applicazione Bigtable.
  • PUBSUB_TOPIC: nome dell'argomento di destinazione Pub/Sub

Passaggi successivi