Modello BigQuery a Bigtable

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

Requisiti della pipeline

• La tabella BigQuery di origine deve esistere.
• La tabella Bigtable deve esistere.
• L'account di servizio worker richiede l'autorizzazione roles/bigquery.datasets.create. Per maggiori informazioni, consulta Introduzione a IAM.

Parametri del modello

Parametri obbligatori

• readIdColumn : il nome della colonna BigQuery in cui è memorizzato 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

• inputTableSpec : la tabella BigQuery da cui leggere. Formato: projectId:datasetId.tablename. 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. Esempio: bigquery-project:dataset.input_table).
• outputDeadletterTable : la tabella BigQuery per i messaggi che non sono riusciti a raggiungere la tabella di output, nel formato <PROJECT_ID>:<DATASET_NAME>.<DEADLETTER_TABLE>. Se non esiste una tabella, viene creata durante l'esecuzione della pipeline. Se non specificato, viene utilizzato <outputTableSpec>_error_records. Ad esempio: id-progetto:set-di-dati.nome-tabella.
• 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, seleziona * da sampledb.sample_table.
• useLegacySql : imposta il valore su true per utilizzare SQL precedente. Questo parametro si applica solo quando si utilizza il parametro query. Il valore predefinito è: false.
• queryLocation : necessaria per la lettura da una visualizzazione autorizzata senza l'autorizzazione della tabella sottostante. (Esempio: Stati Uniti).
• bigtableRpcAttemptTimeoutMs : il timeout in millisecondi per ogni tentativo RPC Bigtable.
• bigtableRpcTimeoutMs : il timeout totale in millisecondi per un'operazione RPC di Bigtable.
• bigtableAdditionalRetryCodes : i codici aggiuntivi per i tentativi. (Esempio: RESOURCE_EXHAUSTED,DEADLINE_EXCEEDED).
• bigtableWriteAppProfile : l'ID del profilo dell'applicazione Bigtable da utilizzare per l'esportazione. Se non specifichi un profilo app, Bigtable utilizza il profilo app 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 : il target di latenza di Bigtable in millisecondi per la limitazione basata sulla latenza.
• bigtableBulkWriteMaxRowKeyCount : numero massimo di chiavi di riga in un'operazione di scrittura batch di Bigtable.
• bigtableBulkWriteMaxRequestSizeBytes : il numero massimo di byte da includere per ogni operazione di scrittura batch Bigtable.

Esegui il modello

Console

1. Vai alla pagina Crea job da modello di Dataflow.
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 BigQuery to Bigtable template.
5. Nei campi dei parametri forniti, inserisci i valori dei parametri.
6. Fai clic su Esegui job.

gcloud

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

Nella shell o nel terminale, esegui il modello:

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/BigQuery_to_Bigtable \
    --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 del progetto Google Cloud in cui vuoi eseguire il job Dataflow
• 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 padre senza data del bucket: gs://dataflow-templates-REGION_NAME/latest/
  • il nome della versione, come 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella padre con data all'interno del bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale data più recente per evitare che queste modifiche che provocano errori influiscano sui flussi di lavoro di produzione.
• READ_COLUMN_ID: la tua colonna di ID univoco di BigQuery.
• INPUT_TABLE_SPEC: il nome della tua tabella BigQuery.
• BIGTABLE_INSTANCE_ID: ID istanza Bigtable.
• BIGTABLE_TABLE_ID: l'ID della tua tabella Bigtable.
• BIGTABLE_COLUMN_FAMILY: la famiglia di colonne della tabella Bigtable.

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",
      "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/BigQuery_to_Bigtable",
   }
}

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
• 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 padre senza data del bucket: gs://dataflow-templates-REGION_NAME/latest/
  • il nome della versione, come 2023-09-12-00_RC00, per utilizzare una versione specifica del modello, che si trova nidificata nella rispettiva cartella padre con data all'interno del bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione: la versione più recente dei modelli potrebbe essere aggiornata con modifiche che provocano errori. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale data più recente per evitare che queste modifiche che provocano errori influiscano sui flussi di lavoro di produzione.
• READ_COLUMN_ID: la tua colonna di ID univoco di BigQuery.
• INPUT_TABLE_SPEC: il nome della tua tabella BigQuery.
• BIGTABLE_INSTANCE_ID: ID istanza Bigtable.
• BIGTABLE_TABLE_ID: l'ID della tua 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.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation.Required;
import org.apache.hadoop.hbase.client.Mutation;

/**
 * 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"},
    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);
  }

  /**
   * 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())
                            .build()))
                .build())
        .apply("WriteToTable", CloudBigtableIO.writeToTable(bigtableTableConfig));

    pipeline.run();
  }
}

Passaggi successivi

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