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 do Bigtable. O modelo pode ler toda a tabela ou ler registros específicos usando uma consulta fornecida.

Requisitos de pipeline

• A tabela de origem do BigQuery precisa existir.
• A tabela do Bigtable precisa existir.
• A conta de serviço do worker precisa da permissão roles/bigquery.datasets.create. Para mais informações, consulte Introdução ao IAM.

Parâmetros do modelo

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 em que será feita a gravação.
• bigtableWriteColumnFamily: o nome do grupo de colunas da tabela do Bigtable em que os dados serão gravados.

Parâmetros opcionais

• timestampColumn: o nome da coluna do BigQuery a ser usada como o carimbo de data/hora da célula da coluna no Bigtable. O valor precisa ter precisão de milissegundos, por exemplo, INT64 / Long. Se uma linha não contiver o campo, o carimbo de data/hora de gravação padrão será usado. A coluna especificada não será incluída como parte da linha no Bigtable como uma coluna separada.
• skipNullValues: flag para indicar se os valores nulos podem ser propagados como um valor vazio ou uma coluna completamente ignorada para aderir ao formato de tabela esparsa do Bigtable. Nos casos em que isso leva a uma linha vazia, por exemplo, uma chave de linha válida, mas sem colunas, a linha não pode ser gravada no Bigtable e será ignorada. O padrão é: falso.
• inputTableSpec: tabela do BigQuery a ser lida. Se você especificar inputTableSpec, o modelo vai ler os dados diretamente do armazenamento do BigQuery usando a API BigQuery Storage Read (https://cloud.google.com/bigquery/docs/reference/storage). Para informações sobre limitações na API Storage Read, consulte https://cloud.google.com/bigquery/docs/reference/storage#limitations. É preciso especificar inputTableSpec ou query. Se você definir os dois parâmetros, o modelo vai usar o query. Exemplo: <BIGQUERY_PROJECT>:<DATASET_NAME>.<INPUT_TABLE>.
• outputDeadletterTable: a tabela do BigQuery para mensagens que não alcançaram a tabela de saída. Se uma tabela não existir, ela será criada durante a execução do pipeline. Se não for especificado, <outputTableSpec>_error_records será usado. Exemplo: <PROJECT_ID>:<DATASET_NAME>.<DEADLETTER_TABLE>.
• consulta: consulta SQL a ser usada para extrair dados do BigQuery. Se o conjunto de dados do BigQuery estiver em um projeto diferente do job do Dataflow, especifique o nome completo do conjunto de dados na consulta SQL. Por exemplo: <PROJECT_ID>.<DATASET_NAME>.<TABLE_NAME>. Por padrão, o parâmetro query usa o GoogleSQL (https://cloud.google.com/bigquery/docs/introduction-sql), a menos que useLegacySql seja true. É preciso especificar inputTableSpec ou query. Se você definir os dois parâmetros, o modelo vai usar o query. Exemplo: select * from sampledb.sample_table.
• useLegacySql: defina como true para usar o SQL legado. Esse parâmetro só se aplica ao usar o parâmetro query. O padrão é false.
• queryLocation: necessário ao ler uma visualização autorizada sem a permissão da tabela subjacente. Exemplo: US.
• queryTempDataset: com essa opção, é possível definir um conjunto de dados existente para criar a tabela temporária para armazenar os resultados da consulta. Exemplo: temp_dataset.
• KMSEncryptionKey: se você estiver lendo do BigQuery usando a origem da consulta, use essa chave do Cloud KMS para criptografar todas as tabelas temporárias criadas. Exemplo: projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key.
• bigtableRpcAttemptTimeoutMs: o tempo limite de cada tentativa de RPC do Bigtable em milissegundos.
• bigtableRpcTimeoutMs: o tempo limite total de uma operação de RPC do Bigtable em milissegundos.
• bigtableAdditionalRetryCodes: os códigos de repetição adicionais. Exemplo: RESOURCE_EXHAUSTED,DEADLINE_EXCEEDED.
• bigtableWriteAppProfile: o ID do perfil do aplicativo do Bigtable a ser usado na exportação. Se você não especificar um perfil de aplicativo, o Bigtable usará o perfil de aplicativo padrão (https://cloud.google.com/bigtable/docs/app-profiles#default-app-profile) da instância.
• bigtableWriteProjectId: o ID do projeto na nuvem do Google Cloud que contém a instância do Bigtable em que os dados serão gravados.
• bigtableBulkWriteLatencyTargetMs: o destino de latência do Bigtable em milissegundos para a limitação com base na latência.
• bigtableBulkWriteMaxRowKeyCount: o número máximo de chaves de linha em uma operação de gravação em lote do Bigtable.
• bigtableBulkWriteMaxRequestSizeBytes: o máximo de bytes a serem incluídos por operação de gravação em lote do Bigtable.
• bigtableBulkWriteFlowControl: quando definido como verdadeiro, ativa o controle de fluxo de gravação em lote, que usa o indicador do servidor para limitar as gravações. O padrão é: falso.

Executar o modelo

Console

1. Acesse a página Criar job usando um modelo do Dataflow.
Acesse Criar job usando um modelo
2. No campo Nome do job, insira um nome exclusivo.
3. Opcional: em Endpoint regional, selecione um valor no menu suspenso. A região padrão é us-central1.

   Para ver uma lista de regiões em que é possível executar um job do Dataflow, consulte Locais do Dataflow.

4. No menu suspenso Modelo do Dataflow, selecione o modelo BigQuery para Bigtable.
5. Nos campos de parâmetro fornecidos, insira os valores de parâmetro.
6. Cliquem em Executar job.

gcloud

Observação: para usar a Google Cloud CLI para executar modelos flex você precisa ter a Google Cloud CLI na versão 284.0.0 ou mais recente.

No 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/ \
    --parameters \
readIdColumn=READ_COLUMN_ID,\
inputTableSpec=INPUT_TABLE_SPEC,\
bigtableWriteInstanceId=BIGTABLE_INSTANCE_ID,\
bigtableWriteTableId=BIGTABLE_TABLE_ID,\
bigtableWriteColumnFamily=BIGTABLE_COLUMN_FAMILY

Substitua:

• PROJECT_ID: o ID do projeto em que você quer executar o job do Dataflow Google Cloud
• JOB_NAME: um nome de job de sua escolha
• REGION_NAME: a região onde você quer implantar o job do Dataflow, por exemplo, us-central1
• VERSION: a versão do modelo que você quer usar

  Use estes valores:

  • latest para usar a versão mais recente do modelo, disponível na pasta mãe não datada no bucket: 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 respectiva pasta mãe datada no bucket: gs://dataflow-templates-REGION_NAME/
  Cuidado: é possível que a versão mais recente dos modelos seja atualizada com alterações interruptivas. Os ambientes de produção precisam usar os modelos mantidos na pasta mãe datada mais recente para evitar que essas alterações interruptivas afetem os fluxos de trabalho de produção.
• READ_COLUMN_ID: coluna de ID exclusivo 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: grupo de colunas da tabela do Bigtable.

API

Para executar o modelo usando a API REST, envie uma solicitação HTTP POST. Para mais informações sobre a API e os respectivos escopos 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/",
   }
}

Substitua:

• PROJECT_ID: o ID do projeto em que você quer executar o job do Dataflow Google Cloud
• JOB_NAME: um nome de job de sua escolha
• LOCATION: a região onde você quer implantar o job do Dataflow, por exemplo, us-central1
• VERSION: a versão do modelo que você quer usar

  Use estes valores:

  • latest para usar a versão mais recente do modelo, disponível na pasta mãe não datada no bucket: 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 respectiva pasta mãe datada no bucket: gs://dataflow-templates-REGION_NAME/
  Cuidado: é possível que a versão mais recente dos modelos seja atualizada com alterações interruptivas. Os ambientes de produção precisam usar os modelos mantidos na pasta mãe datada mais recente para evitar que essas alterações interruptivas afetem os fluxos de trabalho de produção.
• READ_COLUMN_ID: coluna de ID exclusivo 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: grupo 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);
      }
    }
  }
}

A seguir

• Saiba mais sobre os modelos do Dataflow.
• Confira a lista de modelos fornecidos pelo Google.