Modelo do BigQuery para o Clickhouse

Beta

Este recurso está sujeito aos Termos das Soluções Pré-GA na seção Termos Gerais de Serviço dos Termos Específicos de Serviço. Os recursos pré-GA estão disponíveis "no estado em que se encontram" e podem ter suporte limitado. Para mais informações, consulte as descrições da fase de lançamento.

O modelo do BigQuery para o Clickhouse é um pipeline em lote que copia dados de uma tabela do BigQuery para uma tabela do Clickhouse. O modelo pode ler toda a tabela ou ler registros específicos usando uma consulta fornecida.

Requisitos de pipeline

• A tabela do BigQuery de origem precisa existir.
• A tabela do Clickhouse precisa existir.

Parâmetros do modelo

Parâmetros obrigatórios

• jdbcUrl: o URL JDBC do ClickHouse de destino no formato jdbc:clickhouse://host:port/schema. Qualquer opção JDBC pode ser adicionada ao final do URL JDBC. Por exemplo, jdbc:clickhouse://localhost:8123/default.
• clickHouseUsername: o nome de usuário do ClickHouse para autenticação.
• clickHouseTable: o nome da tabela do ClickHouse de destino para inserir os dados.

Parâmetros opcionais

• inputTableSpec: a tabela do BigQuery para leitura. 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. Por exemplo, <BIGQUERY_PROJECT>:<DATASET_NAME>.<INPUT_TABLE>.
• outputDeadletterTable: a tabela do BigQuery para mensagens que não chegaram à 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. Por exemplo, <PROJECT_ID>:<DATASET_NAME>.<DEADLETTER_TABLE>.
• query: a consulta SQL a ser usada para ler 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. Por 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. Por exemplo, US.
• queryTempDataset: com essa opção, é possível definir um conjunto de dados atual para criar a tabela temporária para armazenar os resultados da consulta. Por 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. Por exemplo, projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key.
• clickHousePassword: a senha do ClickHouse para autenticação.
• maxInsertBlockSize: o tamanho máximo do bloco para inserção, se controlarmos a criação de blocos para inserção (opção ClickHouseIO).
• insertDistributedSync: se a configuração estiver ativada, insira a consulta em espera distribuída até que os dados sejam enviados para todos os nós no cluster. (opção ClickHouseIO).
• insertQuorum: para consultas INSERT na tabela replicada, aguarde a gravação do número especificado de réplicas e linearize a adição dos dados. 0: desativado. Essa configuração está desativada nas configurações padrão do servidor (opção ClickHouseIO).
• insertDeduplicate: para consultas INSERT na tabela replicada, especifica que a desduplicação de blocos de inserção precisa ser realizada.
• maxRetries: número máximo de novas tentativas por inserção.

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 Clickhouse.
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/BigQuery_to_Clickhouse \
    --parameters \
jdbcUrl=JDBC_URL,\
clickHouseUsername=CLICKHOUSE_USERNAME,\
clickHouseTable=CLICKHOUSE_TABLE

Substitua:

• PROJECT_ID: o Google Cloud ID do projeto em que você quer executar o job do Dataflow
• JOB_NAME: um nome de job exclusivo de sua escolha
• REGION_NAME: a região em que 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.
• JDBC_URL: seu URL JDBC.
• CLICKHOUSE_USERNAME: seu nome de usuário do Clickhouse.
• CLICKHOUSE_TABLE: sua tabela do Clickhouse.

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": {
          "jdbcUrl": "JDBC_URL",
          "clickHouseUsername": "CLICKHOUSE_USERNAME",
          "clickHouseTable": "CLICKHOUSE_TABLE"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/BigQuery_to_Clickhouse",
   }
}

Substitua:

• PROJECT_ID: o Google Cloud ID do projeto em que você quer executar o job do Dataflow
• JOB_NAME: um nome de job exclusivo de sua escolha
• LOCATION: a região em que 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.
• JDBC_URL: seu URL JDBC.
• CLICKHOUSE_USERNAME: seu nome de usuário do Clickhouse.
• CLICKHOUSE_TABLE: sua tabela do Clickhouse.

Código-fonte do modelo

Java

/*
 * Copyright (C) 2025 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.clickhouse.templates;

import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;

import com.google.api.services.bigquery.model.TableRow;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.clickhouse.options.BigQueryToClickHouseOptions;
import com.google.cloud.teleport.v2.clickhouse.utils.ClickHouseUtils;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.transforms.BigQueryConverters;
import java.util.ArrayList;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.clickhouse.ClickHouseIO;
import org.apache.beam.sdk.io.clickhouse.TableSchema;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.Row;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Template(
    name = "BigQuery_to_ClickHouse",
    category = TemplateCategory.BATCH,
    displayName = "BigQuery to ClickHouse",
    description =
        "The BigQuery to ClickHouse template is a batch pipeline that ingests data from a BigQuery table into ClickHouse table. "
            + "The template can either read the entire table or read specific records using a supplied query.",
    optionsClass = BigQueryToClickHouseOptions.class,
    skipOptions = {
      "javascriptTextTransformReloadIntervalMinutes",
      "pythonExternalTextTransformGcsPath",
      "pythonExternalTextTransformFunctionName"
    },
    flexContainerName = "googlecloud-to-clickhouse",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/bigquery-to-clickhouse",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "The source BigQuery table must exist.",
      "The ClickHouse target table must exist prior running.",
      "This ClickHouse target table must have the exact same column names as the the source table/query."
    })
public class BigQueryToClickHouse {

  private static final Logger log = LoggerFactory.getLogger(BigQueryToClickHouse.class);

  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    BigQueryToClickHouseOptions options =
        PipelineOptionsFactory.fromArgs(args)
            .withValidation()
            .as(BigQueryToClickHouseOptions.class);

    run(options);
  }

  /**
   * Runs the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @return The result of the pipeline execution.
   */
  private static PipelineResult run(BigQueryToClickHouseOptions options) {
    try {
      checkArgument(
          !options.getJdbcUrl().isEmpty(),
          "The ClickHouse JDBC url must have the following template: `jdbc:clickhouse://host:port/schema`");
      checkArgument(
          !options.getClickHouseTable().isEmpty(),
          "ClickHouse target table name is empty. Please provide a valid table name.");

      String clickHouseJDBCURL =
          ClickHouseUtils.setJDBCCredentials(
              options.getJdbcUrl(),
              options.getClickHouseUsername(),
              options.getClickHousePassword());
      TableSchema clickHouseSchema =
          ClickHouseIO.getTableSchema(clickHouseJDBCURL, options.getClickHouseTable());
      Schema beamSchema = TableSchema.getEquivalentSchema(clickHouseSchema);

      // Create the pipeline.
      Pipeline pipeline = Pipeline.create(options);

      /*
       * Step #1: Read from BigQuery. If a query is provided then it is used to get the TableRows.
       */
      PCollection<TableRow> tableRows =
          pipeline.apply(
              "Read From Big Query",
              BigQueryConverters.ReadBigQueryTableRows.newBuilder()
                  .setOptions(options.as(BigQueryToClickHouseOptions.class))
                  .build());

      // Step 2: Transform TableRow to Row
      PCollection<Row> rows =
          tableRows
              .apply(
                  "Convert to Beam Row",
                  ParDo.of(new TableRowToBeamRowFn(beamSchema, clickHouseSchema)))
              .setRowSchema(beamSchema);

      ClickHouseIO.Write clickHouseWriter =
          ClickHouseIO.write(clickHouseJDBCURL, options.getClickHouseTable());

      if (options.getMaxInsertBlockSize() != null) {
        clickHouseWriter.withMaxInsertBlockSize(options.getMaxInsertBlockSize());
      }
      if (options.getInsertDistributedSync() != null) {
        clickHouseWriter.withInsertDistributedSync(options.getInsertDistributedSync());
      }
      if (options.getInsertQuorum() != null) {
        clickHouseWriter.withInsertQuorum(options.getInsertQuorum());
      }
      if (options.getInsertDeduplicate() != null) {
        clickHouseWriter.withInsertDeduplicate(options.getInsertDeduplicate());
      }
      if (options.getMaxRetries() != null) {
        clickHouseWriter.withMaxRetries(options.getMaxRetries());
      }

      // Step 3: Write data to ClickHouse
      rows.apply("Write to ClickHouse", clickHouseWriter);

      return pipeline.run();
    } catch (Exception e) {
      log.error("Error occurred during the BigQuery to ClickHouse template execution: ", e);
      throw new RuntimeException(e);
    }
  }
}

class TableRowToBeamRowFn extends DoFn<TableRow, Row> {
  private Map<String, TableSchema.Column> columnMap;
  private final Schema beamSchema;
  private final TableSchema clickHouseSchema;

  public TableRowToBeamRowFn(Schema beamSchema, TableSchema clickHouseSchema) {
    this.beamSchema = beamSchema;
    this.clickHouseSchema = clickHouseSchema;
  }

  @Setup
  public void setup() {
    columnMap =
        clickHouseSchema.columns().stream()
            .collect(Collectors.toMap(TableSchema.Column::name, column -> column));
  }

  @ProcessElement
  public void processElement(@Element TableRow tableRow, OutputReceiver<Row> out) {
    Row.Builder rowBuilder = Row.withSchema(beamSchema);

    for (Schema.Field field : beamSchema.getFields()) {
      String fieldName = field.getName();
      Object value = tableRow.get(fieldName);
      TableSchema.ColumnType columnType =
          columnMap.get(fieldName) != null ? columnMap.get(fieldName).columnType() : null;

      if (columnType == null) {
        throw new IllegalArgumentException("Couldn't infer type for field: " + fieldName);
      }

      if (value != null) {
        if (columnType.typeName() == TableSchema.ColumnType.FLOAT32.typeName()) {
          rowBuilder.addValue(Float.valueOf(value.toString()));
        } else if (columnType.typeName() == TableSchema.ColumnType.FLOAT64.typeName()) {
          rowBuilder.addValue(Double.valueOf(value.toString()));
        } else if (columnType.typeName() == TableSchema.ColumnType.DATETIME.typeName()
            || columnType.typeName() == TableSchema.ColumnType.DATE.typeName()) {
          rowBuilder.addValue(new DateTime(value.toString()));
        } else if (Objects.equals(columnType.typeName().toString(), "ARRAY")) {
          if (((ArrayList<?>) value).isEmpty()) {
            rowBuilder.addValue(value);
          } else {
            TableSchema.ColumnType finalColumnType = columnType;
            rowBuilder.addValue(
                ((ArrayList<?>) value)
                    .stream()
                        .map(
                            v ->
                                TableSchema.ColumnType.parseDefaultExpression(
                                    finalColumnType.arrayElementType(), v.toString()))
                        .collect(Collectors.toList()));
          }
        } else {
          rowBuilder.addValue(
              TableSchema.ColumnType.parseDefaultExpression(columnType, value.toString()));
        }
      } else {
        rowBuilder.addValue(null);
      }
    }

    Row row = rowBuilder.build();
    out.output(row);
  }
}

A seguir

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