תבנית SQL Server ל-BigQuery

אפשר להשתמש בתבנית SQL Server to BigQuery כדי להעתיק נתונים מטבלאות SQL Server לטבלה ב-BigQuery. הצינור הזה של עיבוד נתונים באצווה עוזר להעביר נתונים יחסיים לניתוח נתונים ולמחסן נתונים ב- Google Cloud. התבנית מתחברת ל-SQL Server באמצעות JDBC. כדי לאבטח פרמטרים רגישים של חיבור כמו שם משתמש, סיסמה ומחרוזת חיבור, אפשר להצפין אותם באמצעות מפתח של Cloud Key Management Service‏ (Cloud KMS). הפרמטרים האלה חייבים לעבור קידוד Base64 לפני ההצפנה. פרטים על הצפנה מופיעים במאמר בנושא הצפנה ופענוח של נתונים באמצעות מפתח סימטרי.

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

• הטבלה ב-BigQuery צריכה להתקיים לפני הפעלת צינור הנתונים.
• לטבלה ב-BigQuery צריכה להיות סכימה תואמת.
• צריכה להיות גישה למסד הנתונים הרלציוני מרשת המשנה שבה פועל Dataflow.

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

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

• ‫connectionURL: מחרוזת כתובת ה-URL של חיבור JDBC. אפשר להעביר אותו כמחרוזת בקידוד Base64 ואז להצפין אותו באמצעות מפתח Cloud KMS, או שהוא יכול להיות סוד של Secret Manager בפורמט projects/{project}/secrets/{secret}/versions/{secret_version}. לדוגמה, jdbc:sqlserver://localhost;databaseName=sampledb.
• ‫outputTable: המיקום של טבלת הפלט ב-BigQuery. לדוגמה, <PROJECT_ID>:<DATASET_NAME>.<TABLE_NAME>.
• ‫bigQueryLoadingTemporaryDirectory: ספריית הזמנית לתהליך הטעינה של BigQuery. לדוגמה, gs://your-bucket/your-files/temp_dir.

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

• ‫connectionProperties: מחרוזת המאפיינים לשימוש בחיבור JDBC. הפורמט של המחרוזת צריך להיות [propertyName=property;]*.מידע נוסף זמין במאמר Configuration Properties (מאפייני הגדרה) (https://dev.mysql.com/doc/connector-j/en/connector-j-reference-configuration-properties.html) במסמכי העזרה של MySQL. לדוגמה, unicode=true;characterEncoding=UTF-8.
• ‫username: שם המשתמש שמשמש לחיבור JDBC. אפשר להעביר אותו כמחרוזת מוצפנת באמצעות מפתח Cloud KMS, או כסוד של Secret Manager בפורמט projects/{project}/secrets/{secret}/versions/{secret_version}.
• ‫password: הסיסמה לשימוש בחיבור JDBC. אפשר להעביר אותו כמחרוזת מוצפנת באמצעות מפתח Cloud KMS, או כסוד של Secret Manager בפורמט projects/{project}/secrets/{secret}/versions/{secret_version}.
• ‫query: השאילתה להרצה במקור כדי לחלץ את הנתונים. שימו לב: יש הבדלים בין חלק מסוגי ה-SQL של JDBC לבין סוגי BigQuery, למרות שהשמות שלהם זהים. כמה מיפויי סוגים חשובים מ-SQL ל-BigQuery שכדאי לזכור הם DATETIME --> TIMESTAMP. יכול להיות שיהיה צורך בהמרת טיפוסים אם הסכימות לא זהות. לדוגמה, select * from sampledb.sample_table.
• ‫KMSEncryptionKey: מפתח ההצפנה של Cloud KMS שבו יש להשתמש כדי לפענח את שם המשתמש, הסיסמה ומחרוזת החיבור. אם מעבירים מפתח Cloud KMS, צריך גם להצפין את שם המשתמש, הסיסמה ומחרוזת החיבור. לדוגמה, projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key.
• ‫useColumnAlias: אם המאפיין הזה מוגדר לערך true, צינור הנתונים משתמש בכינוי העמודה (AS) במקום בשם העמודה כדי למפות את השורות ל-BigQuery. ברירת המחדל היא false.
• ‫isTruncate: אם הערך מוגדר ל-true, צינור הנתונים מבצע חיתוך לפני טעינת הנתונים ל-BigQuery. ברירת המחדל היא false, שגורמת לצינור לצרף נתונים.
• ‫partitionColumn: אם מציינים את partitionColumn יחד עם table, ‏ JdbcIO קורא את הטבלה במקביל על ידי הפעלת כמה מופעים של השאילתה באותה טבלה (תת-שאילתה) באמצעות טווחים. בשלב הזה יש תמיכה בעמודות של מחיצות Long ו-DateTime. העברת סוג העמודה דרך partitionColumnType.
• ‫partitionColumnType: הסוג של partitionColumn. הערכים האפשריים הם long או datetime. ברירת המחדל היא: long.
• ‫table: הטבלה שממנה מתבצעת הקריאה כשמשתמשים במחיצות. הפרמטר הזה מקבל גם שאילתת משנה בסוגריים. לדוגמה, (select id, name from Person) as subq.
• ‫numPartitions: מספר המחיצות. הערך הזה, יחד עם הגבול התחתון והגבול העליון, יוצר צעדים של מחיצות לביטויי פסקה שנוצרו WHERE, שמשמשים לפיצול שווה של עמודת המחיצות. אם הקלט קטן מ-1, המספר מוגדר כ-1.
• ‫lowerBound: הגבול התחתון לשימוש בסכמת המחיצות. אם לא מספקים ערך, Apache Beam מסיק אותו באופן אוטומטי עבור הסוגים הנתמכים. ‫datetime partitionColumnType מקבל ערך של גבול תחתון בפורמט yyyy-MM-dd HH:mm:ss.SSSZ. לדוגמה, 2024-02-20 07:55:45.000+03:30.
• ‫upperBound: הגבול העליון לשימוש בסכמת החלוקה. אם לא מספקים ערך, Apache Beam מסיק אותו באופן אוטומטי עבור הסוגים הנתמכים. ‫datetime partitionColumnType מקבל ערך עליון בפורמט yyyy-MM-dd HH:mm:ss.SSSZ. לדוגמה, 2024-02-20 07:55:45.000+03:30.
• ‫fetchSize: מספר השורות שיש לאחזר מהמסד בכל פעם. לא משמש לקריאות מחולקות. ברירת המחדל היא: 50,000.
• ‫createDisposition: הגדרת CreateDisposition של BigQuery לשימוש. לדוגמה, CREATE_IF_NEEDED או CREATE_NEVER. ברירת המחדל היא: CREATE_NEVER.
• ‫bigQuerySchemaPath: נתיב Cloud Storage של סכימת ה-JSON של BigQuery. אם המדיניות createDisposition מוגדרת לערך CREATE_IF_NEEDED, חובה לציין את הפרמטר הזה. לדוגמה, gs://your-bucket/your-schema.json.
• ‫outputDeadletterTable: הטבלה ב-BigQuery שבה יישמרו הודעות שלא הצליחו להגיע לטבלת הפלט, בפורמט "PROJECT_ID:DATASET_NAME.TABLE_NAME". אם הטבלה לא קיימת, היא נוצרת כשהצינור מופעל. אם לא מציינים את הפרמטר הזה, הצינור ייכשל אם יתרחשו שגיאות כתיבה.אפשר לציין את הפרמטר הזה רק אם הפרמטר useStorageWriteApi או useStorageWriteApiAtLeastOnce מוגדר כ-True.
• ‫disabledAlgorithms: אלגוריתמים שמופרדים בפסיקים וצריך להשבית. אם הערך הזה מוגדר כ-none, אף אלגוריתם לא מושבת. חשוב להשתמש בפרמטר הזה בזהירות, כי יכול להיות שיש אלגוריתמים שמושבתים כברירת מחדל בגלל פגיעויות או בעיות בביצועים. לדוגמה, SSLv3, RC4.
• ‫extraFilesToStage: נתיבים ב-Cloud Storage או סודות ב-Secret Manager של קבצים להעברה זמנית לעובד, מופרדים בפסיקים. הקבצים האלה נשמרים בספרייה ‎ /extra_files בכל עובד. לדוגמה, gs://<BUCKET_NAME>/file.txt,projects/<PROJECT_ID>/secrets/<SECRET_ID>/versions/<VERSION_ID>.
• ‫useStorageWriteApi: אם הערך הוא true, צינור הנתונים משתמש ב-BigQuery Storage Write API‏ (https://cloud.google.com/bigquery/docs/write-api). ערך ברירת המחדל הוא false. מידע נוסף זמין במאמר בנושא שימוש ב-Storage Write API‏ (https://beam.apache.org/documentation/io/built-in/google-bigquery/#storage-write-api).
• ‫useStorageWriteApiAtLeastOnce: כשמשתמשים ב-Storage Write API, המאפיין הזה מציין את סמנטיקת הכתיבה. כדי להשתמש בסמנטיקה של 'לפחות פעם אחת' (https://beam.apache.org/documentation/io/built-in/google-bigquery/#at-least-once-semantics), צריך להגדיר את הפרמטר הזה לערך true. כדי להשתמש בסמנטיקה של מסירה חד-פעמית, מגדירים את הפרמטר לערך false. הפרמטר הזה רלוונטי רק אם הערך של useStorageWriteApi הוא true. ערך ברירת המחדל הוא false.

הרצת התבנית

המסוף

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

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

4. בתפריט הנפתח Dataflow template (תבנית Dataflow), בוחרים בתבנית SQL Server to BigQuery (מ-SQL Server ל-BigQuery).
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/SQLServer_to_BigQuery \
    --parameters \
connectionURL=JDBC_CONNECTION_URL,\
query=SOURCE_SQL_QUERY,\
outputTable=PROJECT_ID:DATASET.TABLE_NAME,
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS,\
connectionProperties=CONNECTION_PROPERTIES,\
username=CONNECTION_USERNAME,\
password=CONNECTION_PASSWORD,\
KMSEncryptionKey=KMS_ENCRYPTION_KEY

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

• ‫JOB_NAME: שם ייחודי של המשימה לפי בחירתכם
• ‫VERSION: הגרסה של התבנית שרוצים להשתמש בה

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

  • latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
  זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרו בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫REGION_NAME: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
• ‫JDBC_CONNECTION_URL: כתובת ה-URL של חיבור JDBC
• ‫SOURCE_SQL_QUERY: שאילתת ה-SQL להרצה במסד הנתונים של המקור
• ‫DATASET: מערך הנתונים שלכם ב-BigQuery
• ‫TABLE_NAME: שם הטבלה ב-BigQuery
• ‫PATH_TO_TEMP_DIR_ON_GCS: הנתיב שלכם ב-Cloud Storage לספריית temp
• ‫CONNECTION_PROPERTIES: מאפייני החיבור של JDBC, אם צריך
• ‫CONNECTION_USERNAME: שם המשתמש של חיבור ה-JDBC
• ‫CONNECTION_PASSWORD: הסיסמה לחיבור JDBC
• ‫KMS_ENCRYPTION_KEY: מפתח ההצפנה של Cloud KMS

API

כדי להפעיל את התבנית באמצעות API בארכיטקטורת REST, שולחים בקשת HTTP POST. מידע נוסף על ה-API ועל היקפי ההרשאות שלו זמין במאמר projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
  "launchParameter": {
    "jobName": "JOB_NAME",
    "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/SQLServer_to_BigQuery",
    "parameters": {
      "connectionURL": "JDBC_CONNECTION_URL",
      "query": "SOURCE_SQL_QUERY",
      "outputTable": "PROJECT_ID:DATASET.TABLE_NAME",
      "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS",
      "connectionProperties": "CONNECTION_PROPERTIES",
      "username": "CONNECTION_USERNAME",
      "password": "CONNECTION_PASSWORD",
      "KMSEncryptionKey":"KMS_ENCRYPTION_KEY"
    },
    "environment": { "zone": "us-central1-f" }
  }
}

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

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

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

  • latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates-REGION_NAME/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית, שאפשר למצוא אותה בתיקיית האב המתאימה עם התאריך בדלי – gs://dataflow-templates-REGION_NAME/
  זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרו בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫LOCATION: האזור שבו רוצים לפרוס את עבודת Dataflow, לדוגמה: us-central1
• ‫JDBC_CONNECTION_URL: כתובת ה-URL של חיבור JDBC
• ‫SOURCE_SQL_QUERY: שאילתת ה-SQL להרצה במסד הנתונים של המקור
• ‫DATASET: מערך הנתונים שלכם ב-BigQuery
• ‫TABLE_NAME: שם הטבלה ב-BigQuery
• ‫PATH_TO_TEMP_DIR_ON_GCS: הנתיב שלכם ב-Cloud Storage לספריית temp
• ‫CONNECTION_PROPERTIES: מאפייני החיבור של JDBC, אם צריך
• ‫CONNECTION_USERNAME: שם המשתמש של חיבור ה-JDBC
• ‫CONNECTION_PASSWORD: הסיסמה לחיבור JDBC
• ‫KMS_ENCRYPTION_KEY: מפתח ההצפנה של Cloud KMS

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

Java

/*
 * Copyright (C) 2018 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.transforms.BigQueryConverters.wrapBigQueryInsertError;
import static com.google.cloud.teleport.v2.utils.GCSUtils.getGcsFileAsString;
import static com.google.cloud.teleport.v2.utils.KMSUtils.maybeDecrypt;

import com.google.api.services.bigquery.model.ErrorProto;
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.coders.FailsafeElementCoder;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.options.JdbcToBigQueryOptions;
import com.google.cloud.teleport.v2.transforms.ErrorConverters;
import com.google.cloud.teleport.v2.utils.BigQueryIOUtils;
import com.google.cloud.teleport.v2.utils.GCSAwareValueProvider;
import com.google.cloud.teleport.v2.utils.JdbcConverters;
import com.google.cloud.teleport.v2.utils.ResourceUtils;
import com.google.cloud.teleport.v2.utils.SecretManagerUtils;
import com.google.cloud.teleport.v2.values.FailsafeElement;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Strings;
import java.util.List;
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.coders.StringUtf8Coder;
import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO.Write;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryInsertError;
import org.apache.beam.sdk.io.gcp.bigquery.InsertRetryPolicy;
import org.apache.beam.sdk.io.gcp.bigquery.TableRowJsonCoder;
import org.apache.beam.sdk.io.gcp.bigquery.WriteResult;
import org.apache.beam.sdk.io.jdbc.JdbcIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider.StaticValueProvider;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TypeDescriptor;
import org.apache.beam.sdk.values.TypeDescriptors;
import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;

/**
 * A template that copies data from a relational database using JDBC to an existing BigQuery table.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/jdbc-and-googlecloud/README_Jdbc_to_BigQuery_Flex.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Jdbc_to_BigQuery_Flex",
    category = TemplateCategory.BATCH,
    displayName = "JDBC to BigQuery with BigQuery Storage API support",
    description = {
      "The JDBC to BigQuery template is a batch pipeline that copies data from a relational database table into an existing BigQuery table. "
          + "This pipeline uses JDBC to connect to the relational database. You can use this template to copy data from any relational database with available JDBC drivers into BigQuery.",
      "For an extra layer of protection, you can also pass in a Cloud KMS key along with a Base64-encoded username, password, and connection string parameters encrypted with the Cloud KMS key. "
          + "See the <a href=\"https://cloud.google.com/kms/docs/reference/rest/v1/projects.locations.keyRings.cryptoKeys/encrypt\">Cloud KMS API encryption endpoint</a> for additional details on encrypting your username, password, and connection string parameters."
    },
    optionsClass = JdbcToBigQueryOptions.class,
    flexContainerName = "jdbc-and-googlecloud",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/jdbc-to-bigquery",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "The JDBC drivers for the relational database must be available.",
      "If BigQuery table already exist before pipeline execution, it must have a compatible schema.",
      "The relational database must be accessible from the subnet where Dataflow runs."
    })
public class JdbcToBigQuery {

  /** Coder for FailsafeElement. */
  private static final FailsafeElementCoder<String, String> FAILSAFE_ELEMENT_CODER =
      FailsafeElementCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of());

  /**
   * Main entry point for executing the pipeline. This will run the pipeline asynchronously. If
   * blocking execution is required, use the {@link JdbcToBigQuery#run} method to start the pipeline
   * and invoke {@code result.waitUntilFinish()} on the {@link PipelineResult}.
   *
   * @param args The command-line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    // Parse the user options passed from the command-line
    JdbcToBigQueryOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(JdbcToBigQueryOptions.class);

    run(options, writeToBQTransform(options));
  }

  /**
   * Create the pipeline with the supplied options.
   *
   * @param options The execution parameters to the pipeline.
   * @param writeToBQ the transform that outputs {@link TableRow}s to BigQuery.
   * @return The result of the pipeline execution.
   */
  @VisibleForTesting
  static PipelineResult run(JdbcToBigQueryOptions options, Write<TableRow> writeToBQ) {
    // Validate BQ STORAGE_WRITE_API options
    BigQueryIOUtils.validateBQStorageApiOptionsBatch(options);
    if (!options.getUseStorageWriteApi()
        && !options.getUseStorageWriteApiAtLeastOnce()
        && !Strings.isNullOrEmpty(options.getOutputDeadletterTable())) {
      throw new IllegalArgumentException(
          "outputDeadletterTable can only be specified if BigQuery Storage Write API is enabled either with useStorageWriteApi or useStorageWriteApiAtLeastOnce.");
    }

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

    /*
     * Steps: 1) Read records via JDBC and convert to TableRow via RowMapper
     *        2) Append TableRow to BigQuery via BigQueryIO
     */
    JdbcIO.DataSourceConfiguration dataSourceConfiguration =
        JdbcIO.DataSourceConfiguration.create(
                StaticValueProvider.of(options.getDriverClassName()),
                maybeDecrypt(
                    maybeParseSecret(options.getConnectionURL()), options.getKMSEncryptionKey()))
            .withUsername(
                maybeDecrypt(
                    maybeParseSecret(options.getUsername()), options.getKMSEncryptionKey()))
            .withPassword(
                maybeDecrypt(
                    maybeParseSecret(options.getPassword()), options.getKMSEncryptionKey()));

    if (options.getDriverJars() != null) {
      dataSourceConfiguration = dataSourceConfiguration.withDriverJars(options.getDriverJars());
    }

    if (options.getConnectionProperties() != null) {
      dataSourceConfiguration =
          dataSourceConfiguration.withConnectionProperties(options.getConnectionProperties());
    }

    /*
     * Step 1: Read records via JDBC and convert to TableRow
     *         via {@link org.apache.beam.sdk.io.jdbc.JdbcIO.RowMapper}
     */
    PCollection<TableRow> rows;
    if (options.getPartitionColumn() != null && options.getTable() != null) {
      // Read with Partitions
      JdbcIO.ReadWithPartitions<TableRow, ?> readIO = null;
      final String partitionColumnType = options.getPartitionColumnType();
      if (partitionColumnType == null || "long".equals(partitionColumnType)) {
        JdbcIO.ReadWithPartitions<TableRow, Long> longTypeReadIO =
            JdbcIO.<TableRow, Long>readWithPartitions(TypeDescriptors.longs())
                .withDataSourceConfiguration(dataSourceConfiguration)
                .withTable(options.getTable())
                .withPartitionColumn(options.getPartitionColumn())
                .withRowMapper(JdbcConverters.getResultSetToTableRow(options.getUseColumnAlias()));
        if (options.getLowerBound() != null && options.getUpperBound() != null) {
          // Check if lower bound and upper bound are long type.
          try {
            longTypeReadIO =
                longTypeReadIO
                    .withLowerBound(Long.valueOf(options.getLowerBound()))
                    .withUpperBound(Long.valueOf(options.getUpperBound()));
          } catch (NumberFormatException e) {
            throw new NumberFormatException(
                "Expected Long values for lowerBound and upperBound, received : " + e.getMessage());
          }
        }
        readIO = longTypeReadIO;
      } else if ("datetime".equals(partitionColumnType)) {
        JdbcIO.ReadWithPartitions<TableRow, DateTime> dateTimeReadIO =
            JdbcIO.<TableRow, DateTime>readWithPartitions(TypeDescriptor.of(DateTime.class))
                .withDataSourceConfiguration(dataSourceConfiguration)
                .withTable(options.getTable())
                .withPartitionColumn(options.getPartitionColumn())
                .withRowMapper(JdbcConverters.getResultSetToTableRow(options.getUseColumnAlias()));
        if (options.getLowerBound() != null && options.getUpperBound() != null) {
          DateTimeFormatter dateFormatter =
              DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss.SSSZ").withOffsetParsed();
          // Check if lowerBound and upperBound are DateTime type.
          try {
            dateTimeReadIO =
                dateTimeReadIO
                    .withLowerBound(dateFormatter.parseDateTime(options.getLowerBound()))
                    .withUpperBound(dateFormatter.parseDateTime(options.getUpperBound()));
          } catch (IllegalArgumentException e) {
            throw new IllegalArgumentException(
                "Expected DateTime values in the format for lowerBound and upperBound, received : "
                    + e.getMessage());
          }
        }
        readIO = dateTimeReadIO;
      } else {
        throw new IllegalStateException("Received unsupported partitionColumnType.");
      }
      if (options.getNumPartitions() != null) {
        readIO = readIO.withNumPartitions(options.getNumPartitions());
      }
      if (options.getFetchSize() != null && options.getFetchSize() > 0) {
        readIO = readIO.withFetchSize(options.getFetchSize());
      }

      rows = pipeline.apply("Read from JDBC with Partitions", readIO);
    } else {
      if (options.getQuery() == null) {
        throw new IllegalArgumentException(
            "Either 'query' or both 'table' AND 'PartitionColumn' must be specified to read from JDBC");
      }
      JdbcIO.Read<TableRow> readIO =
          JdbcIO.<TableRow>read()
              .withDataSourceConfiguration(dataSourceConfiguration)
              .withQuery(new GCSAwareValueProvider(options.getQuery()))
              .withCoder(TableRowJsonCoder.of())
              .withRowMapper(JdbcConverters.getResultSetToTableRow(options.getUseColumnAlias()));

      if (options.getFetchSize() != null && options.getFetchSize() > 0) {
        readIO = readIO.withFetchSize(options.getFetchSize());
      }

      rows = pipeline.apply("Read from JdbcIO", readIO);
    }

    /*
     * Step 2: Append TableRow to an existing BigQuery table
     */
    WriteResult writeResult = rows.apply("Write to BigQuery", writeToBQ);

    /*
     * Step 3.
     * If using Storage Write API, capture failed inserts and either
     *   a) write error rows to DLQ
     *   b) fail the pipeline
     */
    if (options.getUseStorageWriteApi() || options.getUseStorageWriteApiAtLeastOnce()) {
      PCollection<BigQueryInsertError> insertErrors =
          BigQueryIOUtils.writeResultToBigQueryInsertErrors(writeResult, options);

      if (!Strings.isNullOrEmpty(options.getOutputDeadletterTable())) {
        /*
         * Step 3a.
         * Elements that failed inserts into BigQuery are extracted and converted to FailsafeElement
         */
        PCollection<FailsafeElement<String, String>> failedInserts =
            insertErrors
                .apply(
                    "WrapInsertionErrors",
                    MapElements.into(FAILSAFE_ELEMENT_CODER.getEncodedTypeDescriptor())
                        .via((BigQueryInsertError e) -> wrapBigQueryInsertError(e)))
                .setCoder(FAILSAFE_ELEMENT_CODER);

        /*
         * Step 3a Contd.
         * Insert records that failed insert into deadletter table
         */
        failedInserts.apply(
            "WriteFailedRecords",
            ErrorConverters.WriteStringMessageErrors.newBuilder()
                .setErrorRecordsTable(options.getOutputDeadletterTable())
                .setErrorRecordsTableSchema(ResourceUtils.getDeadletterTableSchemaJson())
                .setUseWindowedTimestamp(false)
                .build());
      } else {
        /*
         * Step 3b.
         * Fail pipeline upon write errors if no DLQ was specified
         */
        insertErrors.apply(ParDo.of(new ThrowWriteErrorsDoFn()));
      }
    }

    // Execute the pipeline and return the result.
    return pipeline.run();
  }

  static class ThrowWriteErrorsDoFn extends DoFn<BigQueryInsertError, Void> {
    @ProcessElement
    public void processElement(ProcessContext c) {
      BigQueryInsertError insertError = Objects.requireNonNull(c.element());
      List<String> errorMessages =
          insertError.getError().getErrors().stream()
              .map(ErrorProto::getMessage)
              .collect(Collectors.toList());
      String stackTrace = String.join("\nCaused by:", errorMessages);

      throw new IllegalStateException(
          String.format(
              "Failed to insert row %s.\nCaused by: %s", insertError.getRow(), stackTrace));
    }
  }

  /**
   * Create the {@link Write} transform that outputs the collection to BigQuery as per input option.
   */
  @VisibleForTesting
  static Write<TableRow> writeToBQTransform(JdbcToBigQueryOptions options) {
    // Needed for loading GCS filesystem before Pipeline.Create call
    FileSystems.setDefaultPipelineOptions(options);
    Write<TableRow> write =
        BigQueryIO.writeTableRows()
            .withoutValidation()
            .withCreateDisposition(Write.CreateDisposition.valueOf(options.getCreateDisposition()))
            .withWriteDisposition(
                options.getIsTruncate()
                    ? Write.WriteDisposition.WRITE_TRUNCATE
                    : Write.WriteDisposition.WRITE_APPEND)
            .withCustomGcsTempLocation(
                StaticValueProvider.of(options.getBigQueryLoadingTemporaryDirectory()))
            .withExtendedErrorInfo()
            .to(options.getOutputTable());

    if (Write.CreateDisposition.valueOf(options.getCreateDisposition())
        != Write.CreateDisposition.CREATE_NEVER) {
      write = write.withJsonSchema(getGcsFileAsString(options.getBigQuerySchemaPath()));
    }
    if (options.getUseStorageWriteApi() || options.getUseStorageWriteApiAtLeastOnce()) {
      write = write.withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors());
    }

    return write;
  }

  /**
   * Retrieves a secret value from SecretManagerUtils if the input string matches the specified
   * pattern.
   *
   * @param secret The input string representing a potential secret.
   * @return The secret value if the input matches the pattern and the secret is found, otherwise
   *     the original input string.
   */
  private static String maybeParseSecret(String secret) {
    // Check if the input string is not null.
    if (secret != null) {
      // Check if the input string matches the pattern for secrets stored in SecretManagerUtils.
      if (secret.matches("projects/.*/secrets/.*/versions/.*")) { // Use .* to match any characters
        // Retrieve the secret value from SecretManagerUtils.
        return SecretManagerUtils.getSecret(secret);
      }
    }
    // If the input is null or doesn't match the pattern, return the original input.
    return secret;
  }
}

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

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