Modifiche in tempo reale di Bigtable per il modello Pub/Sub

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

Un flusso di modifiche di Bigtable ti consente di iscriverti alle mutazioni dei dati su base di tabella. Quando ti abboni alle modifiche in tempo reale delle tabelle, si applicano i seguenti vincoli:

  • Vengono restituite 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 fuori sequenza rispetto all'ordinamento dei timestamp dei commit di Bigtable originale.

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

La pipeline richiede che l'argomento Pub/Sub di destinazione esista. L'argomento di destinazione potrebbe essere configurato per convalidare i messaggi utilizzando uno schema. Quando un argomento Pub/Sub specifica uno schema, la pipeline viene avviata solo se lo schema sia valido. A seconda del tipo di schema, utilizza uno dei seguenti schemi Definizioni dell'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 lo schema Protobuf seguente con la 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 dallo stream di modifiche dalla riga corrispondente nella tabella Bigtable. Il modello Pub/Sub appiattisce le voci ogni record di modifiche dei dati in singole modifiche a livello di cella.

Descrizione del messaggio di output Pub/Sub

Nome campo Descrizione
rowKey La chiave riga della riga modificata. Arriva sotto forma di un array di byte. Quando è configurata la codifica dei messaggi JSON, le chiavi di riga vengono restituite come stringhe. Quando viene specificato useBase64Rowkeys, le chiavi di riga sono codificate in Base64. In caso contrario, viene utilizzato un set di caratteri specificato da bigtableChangeStreamCharset 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 mutazione della riga.
column Il qualificatore di colonna interessato dalla mutazione della riga. Per il tipo di mutazione DELETE_FAMILY, il campo della colonna non è impostato. Arriva sotto forma di array di byte. Quando viene configurata la codifica di messaggi JSON, le colonne vengono restituite come stringhe. Quando useBase64ColumnQualifier è specificato, il campo della colonna è con codifica Base64. In caso contrario, viene utilizzato un set di caratteri specificato da bigtableChangeStreamCharset per decodificare i byte della chiave di riga in una stringa.
commitTimestamp Il momento in cui Bigtable applica la mutazione. L'ora è misurata in microsecondi dall'epoca di Unix (1° gennaio 1970 UTC).
timestamp Il valore timestamp della cella interessata dalla mutazione. Per i tipi di mutazione DELETE_CELLS e DELETE_FAMILY, il timestamp non è impostato. Il tempo viene misurato in microsecondi dall'epoca di Unix (1° gennaio 1970 UTC).
timestampFrom Descrive un inizio dell'intervallo di timestamp incluso per tutte le celle eliminate dalla mutazione DELETE_CELLS. Per gli altri tipi di mutazioni, timestampFrom non è impostato. L'ora è misurata in microsecondi dall'epoca di Unix (1° gennaio 1970 UTC).
timestampTo Descrive una fine esclusiva dell'intervallo di timestamp per tutte le celle eliminate dalla mutazione DELETE_CELLS. Per gli altri tipi di mutazione, timestampTo non è impostato.
isGC Un valore booleano che indica se la mutazione è generata da un meccanismo di garbage collection di Bigtable.
tieBreaker Quando due mutazioni vengono registrate contemporaneamente da cluster Bigtable diversi, la mutazione con il valore tiebreaker più alto viene applicata alla tabella di origine. Le mutazioni con valori tiebreaker più bassi vengono ignorate.
value Il nuovo valore impostato dalla mutazione. A meno che non sia impostata l'opzione di pipeline stripValues, il valore viene impostato per le mutazioni SET_CELL. Per gli altri tipi di mutazione, il valore non è impostato. Arriva sotto forma di array di byte. Quando viene configurata la codifica dei messaggi JSON, i valori vengono restituiti come stringhe. Quando useBase64Values è specificato, il valore è codificato Base64. In caso contrario, viene utilizzato un set di caratteri specificato da bigtableChangeStreamCharset per decodificare i byte del valore in una stringa.
sourceInstance Il nome dell'istanza Bigtable in cui è stata registrata la mutazione. Potrebbe verificarsi quando più pipeline eseguono il flusso di 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 uno stream di più pipeline trasferisca modifiche da tabelle diverse allo stesso argomento Pub/Sub.

Requisiti della pipeline

  • L'istanza di origine Bigtable specificata.
  • La tabella di origine Bigtable specificata. Per la tabella devono essere abilitati i flussi di modifiche.
  • Il profilo dell'applicazione Bigtable specificato.
  • L'argomento Pub/Sub specificato deve esistere.

Parametri del modello

Parametri obbligatori

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

Parametri facoltativi

  • messageEncoding: la codifica dei messaggi da pubblicare nell'argomento Pub/Sub. Quando lo schema dell'argomento di destinazione è configurato, la codifica del messaggio è 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 viene configurato lo schema dell'argomento di destinazione, la codifica dei messaggi è 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, column e value del messaggio sono stringhe i cui contenuti sono determinati dalle opzioni della pipeline useBase64Rowkeys, useBase64ColumnQualifiers, useBase64Values e bigtableChangeStreamCharset.
  • stripValues : se impostato su true, le mutazioni di SET_CELL vengono restituite senza l'impostazione di 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 estremamente 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 le riavviate della coda 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. In caso contrario, rowKey viene prodotto 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. In caso contrario, 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 la DLQ. Il valore predefinito è false.
  • bigtableChangeStreamMetadataInstanceId : la modifica di Bigtable trasmette l'ID istanza dei metadati. Il campo predefinito è vuoto.
  • bigtableChangeStreamMetadataTableTableId: l'ID della tabella dei metadati del connettore delle modifiche in tempo reale di Bigtable. Se non viene fornita, una tabella dei metadati del connettore dei flussi di modifiche Bigtable viene creata automaticamente durante l'esecuzione della pipeline. Il campo predefinito è vuoto.
  • bigtableChangeStreamCharset: il nome del set di caratteri degli stream di modifiche Bigtable. Il valore predefinito è UTF-8.
  • bigtableChangeStreamStartTimestamp: il timestamp iniziale (https://tools.ietf.org/html/rfc3339), incluso, da utilizzare per la lettura dei flussi di variazioni. 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 modifiche al nome della famiglia di colonne da ignorare. Il campo predefinito è vuoto.
  • bigtableChangeStreamIgnoreColumns: un elenco separato da virgole di modifiche ai nomi di colonna da ignorare. Il valore predefinito è vuoto.
  • bigtableChangeStreamName: un nome univoco per la pipeline del client. Consente di riprendere l'elaborazione dal momento in cui è stata arrestata una pipeline in esecuzione in precedenza. Il nome predefinito è generato automaticamente. Controlla i log del job Dataflow per il valore utilizzato.
  • bigtableChangeStreamResume: se impostato su true, una nuova pipeline riprende l'elaborazione dal punto in cui si è interrotta una pipeline in esecuzione precedente con lo stesso valore bigtableChangeStreamName. Se la pipeline con il valore bigtableChangeStreamName specificato non è mai stata eseguita, non ne viene avviata una nuova. Se impostato su false, viene avviata una nuova pipeline. Se è già stata eseguita una pipeline con lo stesso valore bigtableChangeStreamName per l'origine specificata, non ne viene avviata una nuova. 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, consulta Località di Dataflow.

  4. Nel menu a discesa Modello di flusso di dati, 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 di tua scelta
  • VERSION: la versione del modello che vuoi utilizzare

    Puoi utilizzare i seguenti valori:

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

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori 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 di 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: l'ID dell'istanza Bigtable.
  • BIGTABLE_TABLE_ID: l'ID della tua tabella Bigtable.
  • BIGTABLE_APPLICATION_PROFILE_ID: l'ID del profilo della tua applicazione Bigtable.
  • PUBSUB_TOPIC: il nome dell'argomento di destinazione Pub/Sub

Passaggi successivi