תבנית BigQuery ל-Clickhouse

בטא

השימוש בתכונה הזו כפוף לתנאי השימוש במוצרים בגרסת טרום-GA, שמפורטים בחלק "תנאי שירות הכלליים" בתנאים הספציפיים של השירות. השימוש בתכונות בגרסת טרום-GA הוא "כפי שהן" (As is), ויכול להיות שהתמיכה בהן תהיה מוגבלת. מידע נוסף זמין בקטע תיאור שלבי ההשקה.

תבנית BigQuery to Clickhouse היא צינור להעברת נתונים באצוות, שמעתיק נתונים מטבלה ב-BigQuery לטבלה קיימת ב-Clickhouse. התבנית יכולה לקרוא את כל הטבלה או לקרוא רשומות ספציפיות באמצעות שאילתה שסופקה.

הדרישות לגבי צינורות עיבוד נתונים

• טבלת המקור ב-BigQuery צריכה להתקיים.
• הטבלה ב-Clickhouse חייבת להתקיים.

פרמטרים של תבניות

פרמטרים נדרשים

• ‫jdbcUrl: כתובת ה-URL של ClickHouse JDBC היעד בפורמט jdbc:clickhouse://host:port/schema. אפשר להוסיף כל אפשרות JDBC בסוף כתובת ה-URL של JDBC. לדוגמה, jdbc:clickhouse://localhost:8123/default.
• ‫clickHouseUsername: שם המשתמש ב-ClickHouse לאימות.
• ‫clickHouseTable: שם טבלת היעד ב-ClickHouse שאליה רוצים להוסיף את הנתונים.

פרמטרים אופציונליים

• ‫inputTableSpec: הטבלה ב-BigQuery שממנה קוראים. אם מציינים את inputTableSpec, התבנית קוראת את הנתונים ישירות מאחסון BigQuery באמצעות BigQuery Storage Read API‏ (https://cloud.google.com/bigquery/docs/reference/storage). מידע על מגבלות ב-Storage Read API זמין בכתובת https://cloud.google.com/bigquery/docs/reference/storage#limitations. חובה לציין inputTableSpec או query. אם מגדירים את שני הפרמטרים, התבנית משתמשת בפרמטר query. לדוגמה, <BIGQUERY_PROJECT>:<DATASET_NAME>.<INPUT_TABLE>.
• ‫outputDeadletterTable: הטבלה ב-BigQuery להודעות שלא הצליחו להגיע לטבלת הפלט. אם טבלה לא קיימת, היא נוצרת במהלך הפעלת צינור הנתונים. אם לא מציינים ערך, המערכת משתמשת בערך <outputTableSpec>_error_records. לדוגמה, <PROJECT_ID>:<DATASET_NAME>.<DEADLETTER_TABLE>.
• ‫query: שאילתת ה-SQL שמשמשת לקריאת נתונים מ-BigQuery. אם מערך הנתונים ב-BigQuery נמצא בפרויקט אחר ממשימת Dataflow, צריך לציין את השם המלא של מערך הנתונים בשאילתת ה-SQL, למשל: <PROJECT_ID>.<DATASET_NAME>.<TABLE_NAME>. כברירת מחדל, הפרמטר query משתמש ב-GoogleSQL (https://cloud.google.com/bigquery/docs/introduction-sql), אלא אם useLegacySql הוא true. חובה לציין inputTableSpec או query. אם מגדירים את שני הפרמטרים, התבנית משתמשת בפרמטר query. לדוגמה, select * from sampledb.sample_table.
• ‫useLegacySql: מגדירים את הערך ל-true כדי להשתמש ב-SQL מדור קודם. הפרמטר הזה רלוונטי רק כשמשתמשים בפרמטר query. ברירת המחדל היא false.
• ‫queryLocation: נדרש כשקוראים מתצוגה מורשית ללא הרשאה לטבלה הבסיסית. לדוגמה, US.
• ‫queryTempDataset: באמצעות האפשרות הזו, אפשר להגדיר מערך נתונים קיים כדי ליצור את הטבלה הזמנית לאחסון תוצאות השאילתה. לדוגמה, temp_dataset.
• ‫KMSEncryptionKey: אם קוראים מ-BigQuery באמצעות מקור שאילתה, משתמשים במפתח Cloud KMS הזה כדי להצפין טבלאות זמניות שנוצרו. לדוגמה, projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key.
• ‫clickHousePassword: הסיסמה של ClickHouse לאימות.
• ‫maxInsertBlockSize: גודל הבלוק המקסימלי להוספה, אם אנחנו שולטים ביצירת הבלוקים להוספה (אפשרות ClickHouseIO).
• ‫insertDistributedSync: אם ההגדרה הזו מופעלת, השאילתה מוכנסת להמתנה מבוזרת עד שהנתונים יישלחו לכל הצמתים באשכול. (אפשרות ClickHouseIO).
• ‫insertQuorum: בשאילתות INSERT בטבלה המשוכפלת, המערכת מחכה עד שהנתונים ייכתבו במספר העותקים שצוין, ואז מוסיפה את הנתונים בצורה לינארית. ‫0 – מושבתת. ההגדרה הזו מושבתת בהגדרות ברירת המחדל של השרת (אפשרות ClickHouseIO).
• ‫insertDeduplicate: בשאילתות INSERT בטבלה המשוכפלת, מציין שצריך לבצע ביטול כפילויות של בלוקים להוספה.
• ‫maxRetries: המספר המקסימלי של ניסיונות חוזרים לכל הוספה.

הרצת התבנית

המסוף

1. עוברים לדף Create job from template (יצירת משימה מתבנית) ב-Dataflow.
כניסה לדף Create job from template
2. בשדה שם המשימה, מזינים שם ייחודי למשימה.
3. אופציונלי: בשדה Regional endpoint (נקודת קצה אזורית), בוחרים ערך מהתפריט הנפתח. אזור ברירת המחדל הוא us-central1.

   רשימת האזורים שבהם אפשר להריץ משימת Dataflow מופיעה במאמר בנושא מיקומי Dataflow.

4. בתפריט הנפתח Dataflow template (תבנית Dataflow), בוחרים את התבנית BigQuery to Clickhouse (מ-BigQuery אל Clickhouse).
5. בשדות הפרמטרים שמופיעים, מזינים את ערכי הפרמטרים.
6. לוחצים על הפעלת העבודה.

gcloud

הערה: כדי להשתמש ב-Google Cloud CLI להפעלת תבניות flex צריך להשתמש בגרסה 284.0.0 או בגרסה מאוחרת יותר של Google Cloud CLI.

במעטפת או בטרמינל, מריצים את התבנית:

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

מחליפים את מה שכתוב בשדות הבאים:

• ‫PROJECT_ID: מזהה הפרויקט שבו רוצים להריץ את משימת Dataflow Google Cloud
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫REGION_NAME: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
• ‫VERSION: הגרסה של התבנית שבה רוצים להשתמש

  אפשר להשתמש בערכים הבאים:

  • ‫latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית האב ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
  זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרו בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫JDBC_URL: כתובת ה-JDBC.
• ‫CLICKHOUSE_USERNAME: שם המשתמש שלכם ב-Clickhouse.
• ‫CLICKHOUSE_TABLE: הטבלה של Clickhouse.

API

כדי להריץ את התבנית באמצעות API בארכיטקטורת REST, שולחים בקשת HTTP POST. מידע נוסף על ה-API ועל היקפי ההרשאות שלו זמין במאמר 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/",
   }
}

מחליפים את מה שכתוב בשדות הבאים:

• ‫PROJECT_ID: מזהה הפרויקט שבו רוצים להריץ את משימת Dataflow Google Cloud
• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫LOCATION: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
• ‫VERSION: הגרסה של התבנית שבה רוצים להשתמש

  אפשר להשתמש בערכים הבאים:

  • ‫latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית האב ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
  זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרו בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫JDBC_URL: כתובת ה-JDBC.
• ‫CLICKHOUSE_USERNAME: שם המשתמש שלכם ב-Clickhouse.
• ‫CLICKHOUSE_TABLE: הטבלה של Clickhouse.

קוד המקור של התבנית

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

המאמרים הבאים

• מידע נוסף על תבניות Dataflow
• כאן אפשר לעיין ברשימת התבניות ש-Google סיפקה.