תבנית BigQuery ל-Bigtable

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

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

• טבלת המקור ב-BigQuery צריכה להתקיים.
• הטבלה ב-Bigtable חייבת להתקיים.
• לחשבון השירות של העובד צריכה להיות ההרשאה roles/bigquery.datasets.create. מידע נוסף זמין במאמר מבוא ל-IAM.

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

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

• ‫readIdColumn: השם של העמודה ב-BigQuery שבה מאוחסן המזהה הייחודי של השורה.
• ‫bigtableWriteInstanceId: המזהה של מכונת Bigtable שמכילה את הטבלה.
• ‫bigtableWriteTableId: המזהה של טבלת Bigtable שאליה רוצים לכתוב.
• ‫bigtableWriteColumnFamily: השם של קבוצת העמודות בטבלת Bigtable שאליה ייכתבו הנתונים.

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

• ‫timestampColumn: השם של העמודה ב-BigQuery שתשמש כחותמת הזמן של התא בעמודה ב-Bigtable. הערך חייב להיות מדויק עד אלפית השנייה, למשל INT64 / Long. אם שורה לא מכילה את השדה, המערכת תשתמש בחותמת הזמן של הכתיבה שמוגדרת כברירת מחדל. העמודה שצוינה לא תיכלל כחלק מהשורה ב-Bigtable כעמודה נפרדת.
• ‫skipNullValues: דגל שמציין אם ערכי null יכולים להתפשט כערך ריק או שהעמודה תדלג לחלוטין כדי להתאים לפורמט של טבלה דלילה ב-Bigtable. במקרים שבהם הפעולה הזו מובילה לשורה ריקה, למשל rowkey תקין אבל ללא עמודות, אי אפשר לכתוב את השורה ב-Bigtable והמערכת תדלג עליה. ברירת המחדל היא: false.
• ‫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.
• ‫bigtableRpcAttemptTimeoutMs: הזמן הקצוב לתפוגה של כל ניסיון RPC ב-Bigtable, באלפיות השנייה.
• ‫bigtableRpcTimeoutMs: הזמן הקצוב הכולל לתפוגת פעולת Bigtable RPC באלפיות השנייה.
• ‫bigtableAdditionalRetryCodes: קודי הניסיון החוזר הנוספים. לדוגמה, RESOURCE_EXHAUSTED,DEADLINE_EXCEEDED.
• ‫bigtableWriteAppProfile: המזהה של פרופיל האפליקציה ב-Bigtable שבו רוצים להשתמש לייצוא. אם לא מציינים פרופיל אפליקציה, Bigtable משתמש בפרופיל האפליקציה שמוגדר כברירת מחדל (https://cloud.google.com/bigtable/docs/app-profiles#default-app-profile) של המופע.
• ‫bigtableWriteProjectId: המזהה של פרויקט הענן ב-Google Cloud שמכיל את מופע Bigtable שאליו ייכתבו הנתונים.
• ‫bigtableBulkWriteLatencyTargetMs: יעד זמן האחזור של Bigtable באלפיות השנייה לצורך הגבלת קצב העברת הנתונים על סמך זמן האחזור.
• ‫bigtableBulkWriteMaxRowKeyCount: המספר המקסימלי של מפתחות שורות בפעולת כתיבה של קבוצת נתונים ב-Bigtable.
• ‫bigtableBulkWriteMaxRequestSizeBytes: מספר הבייטים המקסימלי שייכלל בכל פעולת כתיבה באצווה ב-Bigtable.
• ‫bigtableBulkWriteFlowControl: אם המאפיין מוגדר כ-true, מופעלת בקרה על זרימת נתונים של כתיבה בכמות גדולה, שמשתמשת באות של השרת כדי לווסת נתונים (throttle) את הכתיבות. ברירת המחדל היא: false.

הרצת התבנית

המסוף

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

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

4. בתפריט הנפתח Dataflow template (תבנית Dataflow), בוחרים את התבנית BigQuery to Bigtable (מ-BigQuery ל-Bigtable).
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 \
readIdColumn=READ_COLUMN_ID,\
inputTableSpec=INPUT_TABLE_SPEC,\
bigtableWriteInstanceId=BIGTABLE_INSTANCE_ID,\
bigtableWriteTableId=BIGTABLE_TABLE_ID,\
bigtableWriteColumnFamily=BIGTABLE_COLUMN_FAMILY

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

• ‫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/
  זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרו בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫READ_COLUMN_ID: עמודת המזהה הייחודי ב-BigQuery.
• ‫INPUT_TABLE_SPEC: שם הטבלה ב-BigQuery.
• ‫BIGTABLE_INSTANCE_ID: מזהה מופע Bigtable.
• ‫BIGTABLE_TABLE_ID: מזהה הטבלה ב-Bigtable.
• ‫BIGTABLE_COLUMN_FAMILY: קבוצת העמודות בטבלת Bigtable.

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": {
          "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/",
   }
}

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

• ‫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/
  זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרו בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫READ_COLUMN_ID: עמודת המזהה הייחודי ב-BigQuery.
• ‫INPUT_TABLE_SPEC: שם הטבלה ב-BigQuery.
• ‫BIGTABLE_INSTANCE_ID: מזהה מופע Bigtable.
• ‫BIGTABLE_TABLE_ID: מזהה הטבלה ב-Bigtable.
• ‫BIGTABLE_COLUMN_FAMILY: קבוצת העמודות בטבלת Bigtable.

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

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

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

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