Modello BigQuery a Bigtable

Il modello BigQuery to 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.
• L'account di servizio dell'operatore ha bisogno dell'autorizzazione roles/bigquery.datasets.create. Per maggiori informazioni, 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 contenente 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 di lettura, 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 una tabella non esiste, viene creata durante l'esecuzione della pipeline. Se non specificato, viene utilizzato <outputTableSpec>_error_records. (esempio: your-project-id:your-dataset.your-table-name).
• query : la query SQL da utilizzare per leggere i dati da BigQuery. Se il set di dati BigQuery si trova in un progetto diverso da quello del 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. (Esempio: seleziona * da sampledb.tabella_di_esempio).
• useLegacySql : impostato su true per utilizzare l'SQL precedente. Questo parametro si applica solo se utilizzi il parametro query. Il valore predefinito è false.
• queryLocation : necessario per la lettura 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. (esempio: temp_dataset).
• 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 ripetizione aggiuntivi. (ad es. RESOURCE_EXHAUSTED,DEADLINE_EXCEEDED).
• bigtableWriteAppProfile : l'ID del profilo dell'applicazione Bigtable da utilizzare per l'esportazione. Se non specifichi un profilo dell'app, Bigtable utilizza il profilo dell'app predefinito (https://cloud.google.com/bigtable/docs/app-profiles#default-app-profile) dell'istanza.
• bigtableWriteProjectId : l'ID del progetto Google Cloud contenente l'istanza Bigtable in cui scrivere i dati.
• bigtableBulkWriteLatencyTargetMs : il target di latenza di Bigtable in millisecondi per il throttling in base alla latenza.
• bigtableBulkWriteMaxRowKeyCount : il numero massimo di chiavi di riga in un'operazione di scrittura collettiva Bigtable.
• bigtableBulkWriteMaxRequestSizeBytes : il numero massimo di byte da includere per operazione di scrittura collettiva 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, consulta Località di Dataflow.

4. Nel menu a discesa Modello di flusso di dati, 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 i modelli, devi disporre della versione Google Cloud CLI 284.0.0 o successiva. flex

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 di 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 senza data del 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 rispettiva cartella principale datata nel bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione: la versione più recente dei modelli potrebbe aggiornarsi con modifiche sostanziali. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale più recente datata per impedire che queste modifiche non compatibili influiscano sui flussi di lavoro di produzione.
• READ_COLUMN_ID: la colonna dell'ID univoco BigQuery.
• INPUT_TABLE_SPEC: il nome della tabella BigQuery.
• BIGTABLE_INSTANCE_ID: l'ID 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 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",
      "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 di 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 senza data del 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 rispettiva cartella principale datata nel bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione: la versione più recente dei modelli potrebbe aggiornarsi con modifiche sostanziali. Gli ambienti di produzione devono utilizzare i modelli conservati nella cartella principale più recente datata per impedire che queste modifiche non compatibili influiscano sui flussi di lavoro di produzione.
• READ_COLUMN_ID: la colonna dell'ID univoco BigQuery.
• INPUT_TABLE_SPEC: il nome della tabella BigQuery.
• BIGTABLE_INSTANCE_ID: l'ID 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.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.