Modello BigQuery a Bigtable

Il modello da BigQuery a Bigtable è una pipeline batch che copia i dati da una tabella BigQuery in una tabella Bigtable esistente. Il modello può leggere l'intera tabella o record specifici utilizzando una query fornita.

Requisiti della pipeline

• La tabella BigQuery di origine deve esistere.
• La tabella Bigtable deve esistere.
• Il service account del worker deve disporre dell'autorizzazione roles/bigquery.datasets.create. Per saperne di più, consulta Introduzione a IAM.

Parametri del modello

Parametri obbligatori

• readIdColumn: il nome della colonna BigQuery che memorizza l'identificatore univoco della riga.
• bigtableWriteInstanceId: l'ID dell'istanza Bigtable che contiene la tabella.
• bigtableWriteTableId: l'ID della tabella Bigtable in cui scrivere.
• bigtableWriteColumnFamily: il nome della famiglia di colonne della tabella Bigtable in cui scrivere i dati.

Parametri facoltativi

• timestampColumn: il nome della colonna BigQuery da utilizzare come timestamp per la cella della colonna in Bigtable. Il valore deve avere una precisione in millisecondi, ad es. INT64 / Long. Se una riga non contiene il campo, verrà utilizzato il timestamp di scrittura predefinito. La colonna specificata non verrà inclusa come parte della riga in Bigtable come colonna separata.
• skipNullValues: un flag che indica se i valori null possono essere propagati come valore vuoto o se la colonna viene ignorata completamente per rispettare il formato della tabella sparse di Bigtable. Nei casi in cui ciò comporta una riga vuota, ad es. una rowkey valida, ma nessuna colonna, la riga non può essere scritta in Bigtable e verrà ignorata. Il valore predefinito è false.
• inputTableSpec: la tabella BigQuery da cui leggere. Se specifichi inputTableSpec, il modello legge i dati direttamente dallo spazio di archiviazione BigQuery utilizzando l'API BigQuery Storage Read (https://cloud.google.com/bigquery/docs/reference/storage). Per informazioni sulle limitazioni dell'API Storage Read, consulta https://cloud.google.com/bigquery/docs/reference/storage#limitations. Devi specificare inputTableSpec o query. Se imposti entrambi i parametri, il modello utilizza il parametro query. Ad esempio, <BIGQUERY_PROJECT>:<DATASET_NAME>.<INPUT_TABLE>.
• outputDeadletterTable: la tabella BigQuery per i messaggi che non sono riusciti a raggiungere la tabella di output. Se una tabella non esiste, viene creata durante l'esecuzione della pipeline. Se non viene specificato, viene utilizzato <outputTableSpec>_error_records. Ad esempio, <PROJECT_ID>:<DATASET_NAME>.<DEADLETTER_TABLE>.
• query: la query SQL da utilizzare per leggere i dati da BigQuery. Se il set di dati BigQuery si trova in un progetto diverso dal job Dataflow, specifica il nome completo del set di dati nella query SQL, ad esempio: <PROJECT_ID>.<DATASET_NAME>.<TABLE_NAME>. Per impostazione predefinita, il parametro query utilizza GoogleSQL (https://cloud.google.com/bigquery/docs/introduction-sql), a meno che useLegacySql non sia true. Devi specificare inputTableSpec o query. Se imposti entrambi i parametri, il modello utilizza il parametro query. Ad esempio, select * from sampledb.sample_table.
• useLegacySql: imposta su true per utilizzare SQL precedente. Questo parametro si applica solo quando si utilizza il parametro query. Il valore predefinito è false.
• queryLocation: necessario quando si legge da una vista autorizzata senza l'autorizzazione della tabella sottostante. Ad esempio, US.
• queryTempDataset: con questa opzione, puoi impostare un set di dati esistente per creare la tabella temporanea in cui archiviare i risultati della query. Ad esempio, temp_dataset.
• KMSEncryptionKey: se leggi da BigQuery utilizzando l'origine della query, utilizza questa chiave Cloud KMS per criptare le tabelle temporanee create. Ad esempio, projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key.
• bigtableRpcAttemptTimeoutMs: il timeout per ogni tentativo RPC Bigtable in millisecondi.
• bigtableRpcTimeoutMs: il timeout totale per un'operazione RPC Bigtable in millisecondi.
• bigtableAdditionalRetryCodes: i codici di nuovi tentativi aggiuntivi. Ad esempio, RESOURCE_EXHAUSTED,DEADLINE_EXCEEDED.
• bigtableWriteAppProfile: l'ID del profilo dell'applicazione Bigtable da utilizzare per l'esportazione. Se non specifichi un profilo dell'applicazione, Bigtable utilizza il profilo dell'applicazione predefinito (https://cloud.google.com/bigtable/docs/app-profiles#default-app-profile) dell'istanza.
• bigtableWriteProjectId: l'ID del progetto Google Cloud che contiene l'istanza Bigtable in cui scrivere i dati.
• bigtableBulkWriteLatencyTargetMs: la latenza target di Bigtable in millisecondi per la limitazione basata sulla latenza.
• bigtableBulkWriteMaxRowKeyCount: il numero massimo di rowkey in un'operazione di scrittura batch Bigtable.
• bigtableBulkWriteMaxRequestSizeBytes: il numero massimo di byte da includere per operazione di scrittura batch Bigtable.
• bigtableBulkWriteFlowControl: se impostato su true, abilita il controllo del flusso di scrittura batch che utilizza il segnale del server per limitare le scritture. Il valore predefinito è false.

Esegui il modello

Console

1. Vai alla pagina Dataflow Crea job da modello.
Vai a Crea job da modello
2. Nel campo Nome job, inserisci un nome job univoco.
3. (Facoltativo) Per Endpoint regionale, 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 Dataflow, seleziona il modello Da BigQuery a Bigtable.
5. Inserisci i valori dei parametri nei campi dei parametri forniti.
6. Fai clic su Esegui job.

gcloud

Nota: per utilizzare Google Cloud CLI per eseguire flex i modelli, devi avere la versione 284.0.0 o successiva di Google Cloud CLI.

Esegui il modello nella shell o nel terminale:

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/ \
    --parameters \
readIdColumn=READ_COLUMN_ID,\
inputTableSpec=INPUT_TABLE_SPEC,\
bigtableWriteInstanceId=BIGTABLE_INSTANCE_ID,\
bigtableWriteTableId=BIGTABLE_TABLE_ID,\
bigtableWriteColumnFamily=BIGTABLE_COLUMN_FAMILY

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto in cui vuoi eseguire il job Dataflow Google Cloud
• JOB_NAME: un nome job univoco a tua scelta
• REGION_NAME: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale non datata nel bucket: gs://dataflow-templates-REGION_NAME/latest/
  • il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella cartella principale datata corrispondente nel bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale datata più recente per evitare che queste modifiche che causano interruzioni influiscano sui workflow di produzione.
• READ_COLUMN_ID: la colonna ID univoca di BigQuery.
• INPUT_TABLE_SPEC: il nome della tabella BigQuery.
• BIGTABLE_INSTANCE_ID: l'ID dell'istanza Bigtable.
• BIGTABLE_TABLE_ID: l'ID della tabella Bigtable.
• BIGTABLE_COLUMN_FAMILY: la famiglia di colonne della tabella Bigtable.

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta HTTP POST. Per saperne di più 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",
      "parameters": {
          "readIdColumn": "READ_COLUMN_ID",
          "inputTableSpec": "INPUT_TABLE_SPEC",
          "bigtableWriteInstanceId": "BIGTABLE_INSTANCE_ID",
          "bigtableWriteTableId": "BIGTABLE_TABLE_ID",
          "bigtableWriteColumnFamily": "BIGTABLE_COLUMN_FAMILY"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/",
   }
}

Sostituisci quanto segue:

• PROJECT_ID: l'ID progetto in cui vuoi eseguire il job Dataflow Google Cloud
• JOB_NAME: un nome job univoco a tua scelta
• LOCATION: la regione in cui vuoi eseguire il deployment del job Dataflow, ad esempio us-central1
• VERSION: la versione del modello che vuoi utilizzare

  Puoi utilizzare i seguenti valori:

  • latest per utilizzare la versione più recente del modello, disponibile nella cartella principale non datata nel bucket: gs://dataflow-templates-REGION_NAME/latest/
  • il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella cartella principale datata corrispondente nel bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che causano interruzioni. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale datata più recente per evitare che queste modifiche che causano interruzioni influiscano sui workflow di produzione.
• READ_COLUMN_ID: la colonna ID univoca di BigQuery.
• INPUT_TABLE_SPEC: il nome della tabella BigQuery.
• BIGTABLE_INSTANCE_ID: l'ID dell'istanza Bigtable.
• BIGTABLE_TABLE_ID: l'ID della tabella Bigtable.
• BIGTABLE_COLUMN_FAMILY: la famiglia di colonne della tabella Bigtable.

Codice sorgente del modello

Java

/*
 * Copyright (C) 2021 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.v2.templates;

import static com.google.cloud.teleport.v2.bigtable.utils.BigtableConfig.generateCloudBigtableWriteConfiguration;

import com.google.cloud.bigtable.beam.CloudBigtableIO;
import com.google.cloud.bigtable.beam.CloudBigtableTableConfiguration;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
import com.google.cloud.teleport.v2.bigtable.options.BigtableCommonOptions;
import com.google.cloud.teleport.v2.bigtable.transforms.BigtableConverters;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.templates.BigQueryToBigtable.BigQueryToBigtableOptions;
import com.google.cloud.teleport.v2.transforms.BigQueryConverters;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation.Required;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Dataflow template which reads BigQuery data and writes it to Bigtable. The source data can be
 * either a BigQuery table or an SQL query.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/bigquery-to-bigtable/README_BigQuery_to_Bigtable.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "BigQuery_to_Bigtable",
    category = TemplateCategory.BATCH,
    displayName = "BigQuery to Bigtable",
    description = "A pipeline to export a BigQuery table into Bigtable.",
    optionsClass = BigQueryToBigtableOptions.class,
    optionsOrder = {
      BigQueryToBigtableOptions.class,
      BigQueryConverters.BigQueryReadOptions.class,
      BigtableCommonOptions.class,
      BigtableCommonOptions.WriteOptions.class
    },
    optionalOptions = {"inputTableSpec", "timestampColumn", "skipNullValues"},
    flexContainerName = "bigquery-to-bigtable",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/bigquery-to-bigtable",
    contactInformation = "https://cloud.google.com/support",
    requirements = {
      "The source BigQuery table must exist.",
      "The Bigtable table must exist.",
      "The <a href=\"https://cloud.google.com/dataflow/docs/concepts/security-and-permissions#worker-service-account\">worker service account</a>"
          + " needs the <code>roles/bigquery.datasets.create</code> permission. For"
          + " more information, see <a href=\"https://cloud.google.com/bigquery/docs/access-control\">Introduction to IAM</a>."
    })
public class BigQueryToBigtable {

  /**
   * The {@link BigQueryToBigtableOptions} class provides the custom execution options passed by the
   * executor at the command-line.
   */
  public interface BigQueryToBigtableOptions
      extends BigQueryConverters.BigQueryReadOptions,
          BigtableCommonOptions.WriteOptions,
          GcpOptions {

    @TemplateParameter.Text(
        order = 1,
        regexes = {"[A-Za-z_][A-Za-z_0-9]*"},
        description = "Unique identifier column",
        helpText = "The name of the BigQuery column storing the unique identifier of the row.")
    @Required
    String getReadIdColumn();

    void setReadIdColumn(String value);

    @TemplateParameter.Text(
        order = 2,
        optional = true,
        regexes = {"[A-Za-z_][A-Za-z_0-9]*"},
        description = "Timestamp column identifier",
        helpText =
            "The name of the BigQuery column to be used as the timestamp for the column's cell in Bigtable. The value"
                + " must be millisecond precision, e.g. INT64 / Long. If a row does not contain the field, the default write"
                + " timestamp will be used. The column specified will not be included as part of the row in Bigtable as"
                + " a separate column.")
    @Default.String("")
    String getTimestampColumn();

    void setTimestampColumn(String value);

    @TemplateParameter.Boolean(
        order = 3,
        optional = true,
        description = "Flag to skip null values",
        helpText =
            "Flag to indicate whether nulls may propagate as an empty value or column skipped completely to adhere to "
                + " Bigtable sparse table format. In cases where this leads to an empty row, e.g. a valid rowkey, but no "
                + " columns, the row cannot be written to bigtable and the row will be skipped.")
    @Default.Boolean(false)
    Boolean getSkipNullValues();

    void setSkipNullValues(Boolean value);
  }

  /**
   * Runs a pipeline which reads data from BigQuery and writes it to Bigtable.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    BigQueryToBigtableOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(BigQueryToBigtableOptions.class);

    CloudBigtableTableConfiguration bigtableTableConfig =
        generateCloudBigtableWriteConfiguration(options);

    Pipeline pipeline = Pipeline.create(options);

    pipeline
        .apply(
            "AvroToMutation",
            BigQueryConverters.ReadBigQuery.<Mutation>newBuilder()
                .setOptions(options.as(BigQueryToBigtableOptions.class))
                .setReadFunction(
                    BigQueryIO.read(
                        BigtableConverters.AvroToMutation.newBuilder()
                            .setColumnFamily(options.getBigtableWriteColumnFamily())
                            .setRowkey(options.getReadIdColumn())
                            .setSkipNullValues(options.getSkipNullValues())
                            .setTimestampColumn(options.getTimestampColumn())
                            .build()))
                .build())
        .apply("VerifyAndFilterMutations", ParDo.of((new VerifyAndFilterMutationsFn())))
        .apply("WriteToTable", CloudBigtableIO.writeToTable(bigtableTableConfig));

    pipeline.run();
  }

  /**
   * Filter out invalid Bigtable Mutations, additional validations/filters may be applied e.g. An
   * empty mutation is one that contains no actual cell set.
   */
  static class VerifyAndFilterMutationsFn extends DoFn<Mutation, Mutation> {
    private static final Logger LOG = LoggerFactory.getLogger(VerifyAndFilterMutationsFn.class);

    @ProcessElement
    public void processElement(@Element Mutation mutation, OutputReceiver<Mutation> receiver) {
      if (mutation instanceof Put && mutation.isEmpty()) {
        LOG.warn("Skipping empty mutation for rowkey: {}", Bytes.toStringBinary(mutation.getRow()));
      } else {
        receiver.output(mutation);
      }
    }
  }
}

Passaggi successivi

• Scopri di più sui modelli Dataflow.
• Consulta l'elenco dei modelli forniti da Google.