Modelo do BigQuery para o Clickhouse

Beta

Esta funcionalidade está sujeita aos "Termos das Ofertas de Pré-DG" na secção Termos Gerais de Serviço dos Termos Específicos do Serviço. As funcionalidades de pré-DG estão disponíveis "tal como estão" e podem ter apoio técnico limitado. Para mais informações, consulte as descrições das fases 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 existente. 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 Clickhouse tem de existir.

Parâmetros de modelos

Parâmetros obrigatórios

• jdbcUrl: o URL JDBC do ClickHouse de destino no formato jdbc:clickhouse://host:port/schema. Pode adicionar qualquer opção JDBC no final do URL JDBC. Por exemplo, jdbc:clickhouse://localhost:8123/default.
• clickHouseUsername: o nome de utilizador do ClickHouse para autenticação.
• clickHouseTable: o nome da tabela ClickHouse de destino para inserir os dados.

Parâmetros opcionais

• 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 para armazenar 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.
• clickHousePassword: a palavra-passe 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 definição estiver ativada, insere a consulta em esperas distribuídas 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 escrita para o número especificado de réplicas e linearize a adição dos dados. 0: desativado. Esta definição está desativada nas definições do servidor predefinidas (opção ClickHouseIO).
• insertDeduplicate: para consultas INSERT na tabela replicada, especifica que deve ser feita a desduplicação de blocos de inserção.
• maxRetries: número máximo de novas tentativas por inserção.

Execute o modelo

Consola

1. Aceda à página Dataflow 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 o artigo Localizações do Dataflow.

4. No menu pendente Modelo do fluxo de dados, selecione the BigQuery to Clickhouse 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, tem de ter a versão 284.0.0 ou posterior da CLI do Google Cloud. flex

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_Clickhouse \
    --parameters \
jdbcUrl=JDBC_URL,\
clickHouseUsername=CLICKHOUSE_USERNAME,\
clickHouseTable=CLICKHOUSE_TABLE

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.
• JDBC_URL: o seu URL JDBC.
• CLICKHOUSE_USERNAME: o seu nome de utilizador do Clickhouse.
• CLICKHOUSE_TABLE: a sua tabela do Clickhouse.

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

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.
• JDBC_URL: o seu URL JDBC.
• CLICKHOUSE_USERNAME: o seu nome de utilizador do Clickhouse.
• CLICKHOUSE_TABLE: a 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 = "bigquery-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);
  }
}

O que se segue?

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