Trasmetti le modifiche con Dataflow

Il connettore Beam di Bigtable ti consente di utilizzare Dataflow per leggere i record di modifica dei dati di Bigtable senza dover monitorare o elaborare modifiche alle partizioni nel codice, perché il connettore gestisce la logica per te.

Questo documento descrive come configurare e utilizzare il connettore Beam di Bigtable per leggere un flusso di modifiche utilizzando una pipeline Dataflow. Prima di leggere questo documento, consulta la Panoramica dei flussi di modifiche e familiarizza con Dataflow.

Alternative alla creazione di una pipeline

Se non vuoi creare la tua pipeline Dataflow, puoi utilizzare una delle seguenti opzioni.

Puoi utilizzare un modello Dataflow fornito da Google.

Puoi anche utilizzare gli esempi di codice del tutorial o della guida rapida di Bigtable come punto di partenza per il tuo codice.

Assicurati che il codice che generi utilizzi google cloud libraries-bom versione 26.14.0 o successive.

Dettagli connettore

Il metodo del connettore Beam di Bigtable, BigtableIO.readChangeStream, consente di leggere un flusso di record di modifica dei dati (ChangeStreamMutation) che puoi elaborare. Il connettore Beam di Bigtable è un componente del repository GitHub di Apache Beam. Per una descrizione del codice del connettore, consulta i commenti in BigtableIO.java.

Devi utilizzare il connettore con Beam versione 2.48.0 o successive. Controlla il supporto di runtime di Apache Beam per assicurarti di utilizzare una versione supportata di Java. Poi puoi eseguire il deployment di una pipeline che utilizza il connettore a Dataflow, che gestisce il provisioning e la gestione delle risorse e assiste con la scalabilità e l'affidabilità dell'elaborazione dei dati di flusso.

Per ulteriori informazioni sul modello di programmazione di Apache Beam, consulta la documentazione di Beam.

Raggruppare i dati senza orari degli eventi

I record di modifica dei dati trasmessi in streaming utilizzando il connettore Beam di Bigtable non sono compatibili con le funzioni di Dataflow che dipendono dagli orari degli eventi.

Come spiegato in Replica e filigrane, una filigrana bassa potrebbe non avanzare se la replica della partizione non ha raggiunto il resto dell'istanza. Quando una filigrana bassa smette di avanzare, il flusso di modifiche può bloccarsi.

Per evitare che il flusso si blocchi, il connettore Beam di Bigtable restituisce tutti i dati con un timestamp di output pari a zero. Il timestamp zero fa sì che Dataflow consideri tutti i record di modifica dei dati come dati in ritardo. Di conseguenza, le funzionalità di Dataflow che dipendono dagli orari degli eventi non sono compatibili con i flussi di modifiche in tempo reale di Bigtable. In particolare, non puoi utilizzare le funzioni di finestra, i trigger di orario degli eventi o i timer di orario degli eventi.

In alternativa, puoi utilizzare GlobalWindows con trigger non di orario degli eventi per raggruppare questi dati in ritardo in riquadri, come mostrato nell'esempio del tutorial. Per dettagli su trigger e riquadri, consulta Trigger nella guida alla programmazione di Beam.

Scalabilità automatica

Il connettore supporta la scalabilità automatica di Dataflow, che è abilitata per impostazione predefinita quando si utilizza Runner v2 (obbligatorio). L'algoritmo di scalabilità automatica di Dataflow tiene conto del backlog stimato del flusso di modifiche, che può essere monitorato nella pagina di monitoraggio di Dataflow nella sezione Backlog. Utilizza il flag --maxNumWorkers quando esegui il deployment di un job per limitare il numero di worker.

Per scalare manualmente la pipeline anziché utilizzare la scalabilità automatica, consulta Scalare manualmente una pipeline di flusso.

Limitazioni

Prima di utilizzare il connettore Beam di Bigtable con Dataflow, tieni presente le seguenti limitazioni.

Dataflow Runner v2

Il connettore può essere eseguito solo utilizzando Dataflow Runner v2. Per abilitarlo, specifica --experiments=use_runner_v2 negli argomenti della riga di comando. L'esecuzione con Runner v1 causa l'errore della pipeline con la seguente eccezione:

java.lang.UnsupportedOperationException: BundleFinalizer unsupported by non-portable Dataflow

Snapshot

Il connettore non supporta gli snapshot di Dataflow.

Duplicati

Il connettore Beam di Bigtable trasmette in streaming le modifiche per ogni chiave di riga e ogni cluster in ordine di timestamp di commit, ma poiché a volte riavvia da orari precedenti nel flusso, può produrre duplicati.

Riavvii della pipeline

Se una pipeline Dataflow è stata interrotta per un lungo periodo di tempo, i record di modifica dei dati possono rimanere indietro rispetto al limite di conservazione. Quando la pipeline viene ripresa, Bigtable la interrompe in modo che tu possa avviarne una nuova con un nuovo orario di inizio della richiesta che rientra nel periodo di conservazione. Bigtable esegue questa operazione, anziché avanzare silenziosamente l'orario della richiesta della pipeline originale, per evitare l'eliminazione involontaria dei record di modifica dei dati con timestamp che non rientrano nel periodo di conservazione specificato.

Prima di iniziare

Prima di utilizzare il connettore, completa i seguenti prerequisiti.

Configura l'autenticazione

Per utilizzare gli esempi Java in questa pagina in un ambiente di sviluppo locale, installa e inizializza gcloud CLI, quindi configura Credenziali predefinite dell'applicazione con le tue credenziali utente.

Installa Google Cloud CLI.
Se utilizzi un provider di identità (IdP) esterno, devi prima accedere a gcloud CLI con la tua identità federata.
Se utilizzi una shell locale, crea le credenziali di autenticazione locali per il tuo account utente:
```
gcloud auth application-default login
```
Non devi eseguire questa operazione se utilizzi Cloud Shell.

Se viene restituito un errore di autenticazione e utilizzi un provider di identità (IdP) esterno, verifica di aver acceduto a gcloud CLI con la tua identità federata.

Per saperne di più, consulta Configura l'autenticazione per un ambiente di sviluppo locale.

Per informazioni sulla configurazione dell'autenticazione per un ambiente di produzione, consulta Configura le credenziali predefinite dell'applicazione per il codice in esecuzione su Google Cloud .

Abilita un flusso di modifiche

Prima di poter leggere una tabella, devi abilitare un flusso di modifiche. Puoi anche creare una nuova tabella con le modifiche in tempo reale abilitate.

Tabella dei metadati del flusso di modifiche

Quando trasmetti in streaming le modifiche con Dataflow, il connettore Beam di Bigtable crea una tabella di metadati denominata __change_stream_md_table per impostazione predefinita. La tabella dei metadati del flusso di modifiche gestisce lo stato operativo del connettore e archivia i metadati relativi ai record di modifica dei dati.

Per impostazione predefinita, il connettore crea la tabella nella stessa istanza della tabella di cui viene eseguito lo streaming. Per garantire il corretto funzionamento della tabella, il profilo dell'app per la tabella dei metadati deve utilizzare il routing a cluster singolo e avere le transazioni a riga singola abilitate.

Per ulteriori informazioni sulla trasmissione in streaming delle modifiche da Bigtable con il connettore Beam di Bigtable, consulta la documentazione di BigtableIO.

Ruoli obbligatori

Per ottenere le autorizzazioni necessarie per leggere un flusso di modifiche di Bigtable utilizzando Dataflow, chiedi all'amministratore di concederti i seguenti ruoli IAM.

Per leggere le modifiche da Bigtable, devi disporre di questo ruolo:

Amministratore Bigtable (roles/bigtable.admin) nell'istanza Bigtable che contiene la tabella da cui prevedi di trasmettere in streaming le modifiche

Per eseguire il job Dataflow, devi disporre di questi ruoli:

Dataflow Developer (roles/dataflow.developer) nel progetto contenente le risorse Cloud
Dataflow Worker (roles/dataflow.worker) nel progetto contenente le risorse Cloud
Storage Object Admin (roles/storage.objectAdmin) nei bucket Cloud Storage che prevedi di utilizzare

Per saperne di più sulla concessione dei ruoli, consulta Gestisci l'accesso.

Potresti anche riuscire a ottenere le autorizzazioni richieste tramite ruoli personalizzati o altri ruoli predefiniti.

Aggiungi il connettore Beam di Bigtable come dipendenza

Aggiungi al file pom.xml di Maven un codice simile alla seguente dipendenza. La versione deve essere 2.48.0 o successive.

<dependencies>
  <dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
    <version>VERSION</version>
  </dependency>
</dependencies>

Leggi il flusso di modifiche

Per creare una pipeline Dataflow per leggere i record di modifica dei dati, configura il connettore, quindi aggiungi trasformazioni e sink. Poi utilizzi il connettore per leggere gli oggetti ChangeStreamMutation in una pipeline Beam.

Gli esempi di codice in questa sezione, scritti in Java, mostrano come creare una pipeline e utilizzarla per convertire le coppie chiave-valore in una stringa. Ogni coppia è costituita da una chiave di riga e da un oggetto ChangeStreamMutation. La pipeline converte le voci di ogni oggetto in una stringa separata da virgole.

Crea la pipeline

Questo esempio di codice Java mostra come creare la pipeline:

BigtableOptions options =
    PipelineOptionsFactory.fromArgs(args).withValidation().as(BigtableOptions.class);
Pipeline p = Pipeline.create(options);

final Instant startTime = Instant.now();

p.apply(
        "Read Change Stream",
        BigtableIO.readChangeStream()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId())
            .withAppProfileId(options.getBigtableAppProfile())
            .withStartTime(startTime))
    .apply(
        "Flatten Mutation Entries",
        FlatMapElements.into(TypeDescriptors.strings())
            .via(ChangeStreamsHelloWorld::mutationEntriesToString))
    .apply(
        "Print mutations",
        ParDo.of(
            new DoFn<String, Void>() { // a DoFn as an anonymous inner class instance
              @ProcessElement
              public void processElement(@Element String mutation) {
                System.out.println("Change captured: " + mutation);
              }
            }));
p.run();

Elabora i record di modifica dei dati

Questo esempio mostra come scorrere tutte le voci in un record di modifica dei dati per una riga e chiamare un metodo di conversione in stringa in base al tipo di voce.

Per un elenco dei tipi di voci che un record di modifica dei dati può contenere, consulta Contenuto di un record di modifica dei dati.

static List<String> mutationEntriesToString(KV<ByteString, ChangeStreamMutation> mutationPair) {
  List<String> mutations = new ArrayList<>();
  String rowKey = mutationPair.getKey().toStringUtf8();
  ChangeStreamMutation mutation = mutationPair.getValue();
  MutationType mutationType = mutation.getType();
  for (Entry entry : mutation.getEntries()) {
    if (entry instanceof SetCell) {
      mutations.add(setCellToString(rowKey, mutationType, (SetCell) entry));
    } else if (entry instanceof DeleteCells) {
      mutations.add(deleteCellsToString(rowKey, mutationType, (DeleteCells) entry));
    } else if (entry instanceof DeleteFamily) {
      // Note: DeleteRow mutations are mapped into one DeleteFamily per-family
      mutations.add(deleteFamilyToString(rowKey, mutationType, (DeleteFamily) entry));
    } else {
      throw new RuntimeException("Entry type not supported.");
    }
  }
  return mutations;
}

In questo esempio viene convertita una voce di scrittura:

private static String setCellToString(String rowKey, MutationType mutationType, SetCell setCell) {
  List<String> mutationParts =
      Arrays.asList(
          rowKey,
          mutationType.name(),
          "SetCell",
          setCell.getFamilyName(),
          setCell.getQualifier().toStringUtf8(),
          setCell.getValue().toStringUtf8());
  return String.join(",", mutationParts);
}

In questo esempio viene convertita una voce di eliminazione delle celle:

private static String deleteCellsToString(
    String rowKey, MutationType mutationType, DeleteCells deleteCells) {
  String timestampRange =
      deleteCells.getTimestampRange().getStart() + "-" + deleteCells.getTimestampRange().getEnd();
  List<String> mutationParts =
      Arrays.asList(
          rowKey,
          mutationType.name(),
          "DeleteCells",
          deleteCells.getFamilyName(),
          deleteCells.getQualifier().toStringUtf8(),
          timestampRange);
  return String.join(",", mutationParts);
}

In questo esempio viene convertita una voce di eliminazione di una famiglia di colonne:


private static String deleteFamilyToString(
    String rowKey, MutationType mutationType, DeleteFamily deleteFamily) {
  List<String> mutationParts =
      Arrays.asList(rowKey, mutationType.name(), "DeleteFamily", deleteFamily.getFamilyName());
  return String.join(",", mutationParts);
}

Monitoraggio

Le seguenti risorse nella Google Cloud console ti consentono di monitorare le tue Google Cloud risorse durante l'esecuzione di una pipeline Dataflow per leggere un flusso di modifiche di Bigtable:

In particolare, controlla le seguenti metriche:

Nella pagina delle informazioni dettagliate sul sistema Bigtable, controlla le seguenti metriche:
- Dati Utilizzo della CPU per flussi di modifiche in tempo reale nella metrica cpu_load_by_app_profile_by_method_by_table. Mostra l'impatto del flusso di modifiche sull'utilizzo della CPU del cluster.
- Utilizzo dello spazio di archiviazione per le modifiche in tempo reale (byte) (change_stream_log_used_bytes).
Nella pagina di monitoraggio di Dataflow, controlla l'aggiornamento dei dati. Questa metrica mostra la differenza tra l'ora attuale e la filigrana, che è di circa due minuti, con picchi occasionali di uno o due minuti in più. L'aggiornamento dei dati non indica se i record di modifica dei dati vengono elaborati lentamente. Per garantire la salute e le prestazioni continue delle applicazioni critiche, monitora la metrica di aggiornamento dei dati di Dataflow ed esegui le seguenti azioni:
- Se la metrica di aggiornamento dei dati è costantemente superiore alla soglia, è possibile che la pipeline non abbia risorse sufficienti. Ti consigliamo di aggiungere altri worker Dataflow.
- Se i worker Dataflow sono ben sottoposti a provisioning, ma l'aggiornamento dei dati è in aumento o è costantemente elevato, contatta Google Cloud l'assistenza.
La metrica processing_delay_from_commit_timestamp_MEAN di Dataflow può indicare il tempo di elaborazione medio dei record di modifica dei dati durante la durata del job.

La metrica server/latencies di Bigtable non è utile quando monitori una pipeline Dataflow che legge un flusso di modifiche di Bigtable, perché riflette la durata della richiesta di streaming, non la latenza di elaborazione dei record di modifica dei dati. Una latenza elevata in un flusso di modifiche non significa che le richieste vengano elaborate lentamente, ma che la connessione è rimasta aperta per quel periodo di tempo.