Modelo do BigQuery para o Bigtable

O modelo do BigQuery para o Bigtable é um pipeline em lote que copia dados de uma tabela do BigQuery para uma tabela existente do Bigtable. O modelo pode ler toda a tabela ou ler registos específicos através de uma consulta fornecida.

Requisitos do pipeline

• A tabela de origem do BigQuery tem de existir.
• A tabela do Bigtable tem de existir.
• A conta de serviço do trabalhador precisa da autorização roles/bigquery.datasets.create. Para mais informações, consulte o artigo Introdução à IAM.

Parâmetros de modelos

Parâmetros obrigatórios

• readIdColumn: o nome da coluna do BigQuery que armazena o identificador exclusivo da linha.
• bigtableWriteInstanceId: o ID da instância do Bigtable que contém a tabela.
• bigtableWriteTableId: o ID da tabela do Bigtable na qual escrever.
• bigtableWriteColumnFamily: o nome da família de colunas da tabela do Bigtable na qual escrever dados.

Parâmetros opcionais

• timestampColumn: o nome da coluna do BigQuery a usar como a data/hora da célula da coluna no Bigtable. O valor tem de ter uma precisão de milissegundos, por exemplo, INT64 / Long. Se uma linha não contiver o campo, é usada a data/hora de gravação predefinida. A coluna especificada não é incluída como parte da linha no Bigtable como uma coluna separada.
• skipNullValues: indicador para indicar se os valores nulos podem ser propagados como um valor vazio ou se a coluna é completamente ignorada para aderir ao formato de tabela esparsa do Bigtable. Nos casos em que isto leva a uma linha vazia, por exemplo, uma rowkey válida, mas sem colunas, a linha não pode ser escrita no BigTable e é ignorada. A predefinição é: false.
• inputTableSpec: a tabela do BigQuery a partir da qual ler. Se especificar inputTableSpec, o modelo lê os dados diretamente do armazenamento do BigQuery através da API BigQuery Storage Read (https://cloud.google.com/bigquery/docs/reference/storage). Para obter informações sobre as limitações na API Storage Read, consulte https://cloud.google.com/bigquery/docs/reference/storage#limitations. Tem de especificar inputTableSpec ou query. Se definir ambos os parâmetros, o modelo usa o parâmetro query. Por exemplo, <BIGQUERY_PROJECT>:<DATASET_NAME>.<INPUT_TABLE>.
• outputDeadletterTable: a tabela do BigQuery para mensagens que não conseguiram alcançar a tabela de saída. Se uma tabela não existir, é criada durante a execução do pipeline. Se não for especificado, é usado <outputTableSpec>_error_records. Por exemplo, <PROJECT_ID>:<DATASET_NAME>.<DEADLETTER_TABLE>.
• query: a consulta SQL a usar para ler dados do BigQuery. Se o conjunto de dados do BigQuery estiver num projeto diferente do trabalho do Dataflow, especifique o nome completo do conjunto de dados na consulta SQL, por exemplo: <PROJECT_ID>.<DATASET_NAME>.<TABLE_NAME>. Por predefinição, o parâmetro query usa o GoogleSQL (https://cloud.google.com/bigquery/docs/introduction-sql), a menos que useLegacySql seja true. Tem de especificar inputTableSpec ou query. Se definir ambos os parâmetros, o modelo usa o parâmetro query. Por exemplo, select * from sampledb.sample_table.
• useLegacySql: definido como true para usar o SQL antigo. Este parâmetro só se aplica quando usa o parâmetro query. A predefinição é false.
• queryLocation: necessário quando lê a partir de uma vista autorizada sem a autorização da tabela subjacente. Por exemplo, US.
• queryTempDataset: com esta opção, pode definir um conjunto de dados existente para criar a tabela temporária que armazena os resultados da consulta. Por exemplo, temp_dataset.
• KMSEncryptionKey: se estiver a ler a partir do BigQuery através da origem de consulta, use esta chave do Cloud KMS para encriptar todas as tabelas temporárias criadas. Por exemplo, projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key.
• bigtableRpcAttemptTimeoutMs: o limite de tempo para cada tentativa de RPC do Bigtable em milissegundos.
• bigtableRpcTimeoutMs: o limite de tempo total para uma operação RPC do Bigtable em milissegundos.
• bigtableAdditionalRetryCodes: os códigos de repetição adicionais. Por exemplo, RESOURCE_EXHAUSTED,DEADLINE_EXCEEDED.
• bigtableWriteAppProfile: o ID do perfil da aplicação do Bigtable a usar para a exportação. Se não especificar um perfil de app, o Bigtable usa o perfil de app predefinido (https://cloud.google.com/bigtable/docs/app-profiles#default-app-profile) da instância.
• bigtableWriteProjectId: o ID do projeto do Google Cloud que contém a instância do Bigtable para escrever dados.
• bigtableBulkWriteLatencyTargetMs: o objetivo de latência do Bigtable em milissegundos para a limitação baseada na latência.
• bigtableBulkWriteMaxRowKeyCount: o número máximo de chaves de linhas numa operação de escrita em lote do Bigtable.
• bigtableBulkWriteMaxRequestSizeBytes: o número máximo de bytes a incluir por operação de escrita em lote do Bigtable.

Execute o modelo

Consola

1. Aceda à página do fluxo de dados Criar tarefa a partir de um modelo.
Aceda a Criar tarefa a partir de modelo
2. No campo Nome da tarefa, introduza um nome exclusivo para a tarefa.
3. Opcional: para Ponto final regional, selecione um valor no menu pendente. A região predefinida é us-central1.

   Para ver uma lista das regiões onde pode executar uma tarefa do Dataflow, consulte as localizações do Dataflow.

4. No menu pendente Modelo do fluxo de dados, selecione the BigQuery to Bigtable template.
5. Nos campos de parâmetros fornecidos, introduza os valores dos parâmetros.
6. Clique em Executar tarefa.

gcloud

Nota: para usar a CLI do Google Cloud para executar modelos flex, tem de ter a versão CLI do Google Cloud 284.0.0 ou posterior.

Na shell ou no terminal, execute o modelo:

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

Substitua o seguinte:

• PROJECT_ID: o ID do projeto onde quer executar a tarefa do Dataflow Google Cloud
• JOB_NAME: um nome de tarefa exclusivo à sua escolha
• REGION_NAME: a região onde quer implementar a tarefa do Dataflow, por exemplo, us-central1
• VERSION: a versão do modelo que quer usar

  Pode usar os seguintes valores:

  • latest para usar a versão mais recente do modelo, que está disponível na pasta principal sem data no contentor: gs://dataflow-templates-REGION_NAME/latest/
  • o nome da versão, como 2023-09-12-00_RC00, para usar uma versão específica do modelo, que pode ser encontrada aninhada na pasta principal com a data correspondente no contentor: gs://dataflow-templates-REGION_NAME/
  Atenção: a versão mais recente dos modelos pode ser atualizada com alterações destrutivas. Os seus ambientes de produção devem usar modelos mantidos na pasta principal datada mais recente para evitar que estas alterações significativas afetem os seus fluxos de trabalho de produção.
• READ_COLUMN_ID: a coluna de ID único do BigQuery.
• INPUT_TABLE_SPEC: o nome da tabela do BigQuery.
• BIGTABLE_INSTANCE_ID: o ID da instância do Bigtable.
• BIGTABLE_TABLE_ID: o ID da tabela do Bigtable.
• BIGTABLE_COLUMN_FAMILY: a família de colunas da tabela do Bigtable.

API

Para executar o modelo através da API REST, envie um pedido HTTP POST. Para mais informações sobre a API e os respetivos âmbitos de autorização, consulte 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",
   }
}

Substitua o seguinte:

• PROJECT_ID: o ID do projeto onde quer executar a tarefa do Dataflow Google Cloud
• JOB_NAME: um nome de tarefa exclusivo à sua escolha
• LOCATION: a região onde quer implementar a tarefa do Dataflow, por exemplo, us-central1
• VERSION: a versão do modelo que quer usar

  Pode usar os seguintes valores:

  • latest para usar a versão mais recente do modelo, que está disponível na pasta principal sem data no contentor: gs://dataflow-templates-REGION_NAME/latest/
  • o nome da versão, como 2023-09-12-00_RC00, para usar uma versão específica do modelo, que pode ser encontrada aninhada na pasta principal com a data correspondente no contentor: gs://dataflow-templates-REGION_NAME/
  Atenção: a versão mais recente dos modelos pode ser atualizada com alterações destrutivas. Os seus ambientes de produção devem usar modelos mantidos na pasta principal datada mais recente para evitar que estas alterações significativas afetem os seus fluxos de trabalho de produção.
• READ_COLUMN_ID: a coluna de ID único do BigQuery.
• INPUT_TABLE_SPEC: o nome da tabela do BigQuery.
• BIGTABLE_INSTANCE_ID: o ID da instância do Bigtable.
• BIGTABLE_TABLE_ID: o ID da tabela do Bigtable.
• BIGTABLE_COLUMN_FAMILY: a família de colunas da tabela do Bigtable.

Código-fonte do modelo

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);
      }
    }
  }
}

O que se segue?

• Saiba mais sobre os modelos do Dataflow.
• Consulte a lista de modelos fornecidos pela Google.