Modello da 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.
• 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 è archiviato 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 su true per utilizzare l'SQL precedente. Questo parametro si applica solo quando si utilizza il parametro query. Il valore predefinito è false.
• queryLocation : necessaria durante la lettura da una vista autorizzata senza l'autorizzazione della tabella sottostante. (ad esempio US).
• bigtableRpcAttemptTimeoutMs: il timeout per ogni tentativo RPC Bigtable in millisecondi.
• bigtableRpcTimeoutMs: il timeout totale per un'operazione RPC Bigtable in millisecondi.
• bigtableAdditionalRetryCodes : i codici aggiuntivi per i nuovi tentativi. 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'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 ogni operazione di scrittura batch di 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. Il valore predefinito è us-central1.

   Per un elenco di regioni in cui è possibile eseguire un job Dataflow, consulta Località di Dataflow.

4. Dal 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 devi avere la versione Google Cloud CLI 284,0,0 o in un secondo momento.

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 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 essere aggiornata con interruzioni modifiche. I tuoi ambienti di produzione devono utilizzare i modelli conservati nel più recente dated (datata) per evitare che queste modifiche che provocano errori influiscano su per i 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 dell'istanza Bigtable.
• BIGTABLE_TABLE_ID: l'ID della tua tabella Bigtable.
• BIGTABLE_COLUMN_FAMILY: la tua famiglia di colonne delle tabelle Bigtable.

API

Per eseguire il modello utilizzando l'API REST, invia una richiesta POST HTTP. Per ulteriori informazioni sul API e i 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 progetto Google Cloud in cui vuoi eseguire il job Dataflow
• JOB_NAME: un nome job univoco di tua scelta
• LOCATION: la regione in cui vuoi di 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 in cartella principale non-dated nel bucket: gs://dataflow-templates-REGION_NAME/latest/
  • il nome della versione, ad esempio 2023-09-12-00_RC00, per utilizzare una versione specifica , che puoi trovare nidificata nella rispettiva cartella principale con data del bucket: gs://dataflow-templates-REGION_NAME/
  Attenzione:la versione più recente dei modelli potrebbe essere aggiornata con interruzioni modifiche. 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 dell'istanza Bigtable.
• BIGTABLE_TABLE_ID: l'ID della tua tabella Bigtable.
• BIGTABLE_COLUMN_FAMILY: la tua famiglia di colonne delle tabelle 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.