תבנית Java Database Connectivity ‏ (JDBC) ל-BigQuery

תבנית ה-JDBC ל-BigQuery היא צינור (pipeline) של אצווה שמעתיק נתונים מטבלת מסד נתונים יחסי לטבלה קיימת ב-BigQuery. הצינור הזה משתמש ב-JDBC כדי להתחבר למסד הנתונים הרלציוני. אפשר להשתמש בתבנית הזו כדי להעתיק נתונים מכל מסד נתונים רלציוני עם מנהלי התקנים זמינים של JDBC אל BigQuery.

כדי להוסיף עוד שכבת הגנה, אפשר להעביר מפתח Cloud KMS, יחד עם פרמטרים של שם משתמש, סיסמה ומחרוזת חיבור בהצפנת Base64, שמוצפנים באמצעות מפתח Cloud KMS. פרטים נוספים על הצפנה של שם המשתמש, הסיסמה והפרמטרים של מחרוזת החיבור זמינים במאמר נקודת הקצה של הצפנת Cloud KMS API.

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

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

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

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

• ‫driverJars: רשימה מופרדת בפסיקים של קובצי JAR של מנהלי התקנים. לדוגמה, gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar.
• ‫driverClassName: שם המחלקה של מנהל ההתקן של JDBC. לדוגמה, com.mysql.jdbc.Driver.
• ‫connectionURL: מחרוזת כתובת ה-URL של חיבור JDBC. לדוגמה, jdbc:mysql://some-host:3306/sampledb. אפשר להעביר את הערך הזה כמחרוזת שמוצפנת באמצעות מפתח Cloud KMS ואז מקודדת ב-Base64. מסירים את התווים של הרווחים מהמחרוזת שמקודדת ב-Base64. שימו לב להבדל בין מחרוזת חיבור למסד נתונים של Oracle שאינו RAC‏ (jdbc:oracle:thin:@some-host:<port>:<sid>) לבין מחרוזת חיבור למסד נתונים של Oracle RAC‏ (jdbc:oracle:thin:@//some-host[:<port>]/<service_name>). לדוגמה, jdbc:mysql://some-host:3306/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: השאילתה להרצה במקור כדי לחלץ את הנתונים. שימו לב: יש הבדלים בין חלק מהסוגים של JDBC SQL ו-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: הגדרת BigQuery CreateDisposition לשימוש. לדוגמה, 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.

הפעלת התבנית

הערה: אם אתם מחפשים הוראות להפעלת תבנית JDBC ל-BigQuery מדור קודם (קלאסי), כדאי לעיין במסמכי התבנית JDBC ל-BigQuery (מדורה קודם).

המסוף

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

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

4. בתפריט הנפתח Dataflow template (תבנית של העברת נתונים), בוחרים באפשרות the JDBC to BigQuery with BigQuery Storage API support template.
5. בשדות הפרמטרים שמופיעים, מזינים את ערכי הפרמטרים.
6. לוחצים על הפעלת העבודה.

gcloud

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

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

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Jdbc_to_BigQuery_Flex \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --parameters \
       driverJars=DRIVER_JARS,\
       driverClassName=DRIVER_CLASS_NAME,\
       connectionURL=CONNECTION_URL,\
       outputTable=OUTPUT_TABLE,\
       bigQueryLoadingTemporaryDirectory=BIG_QUERY_LOADING_TEMPORARY_DIRECTORY,\

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

• ‫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
• ‫DRIVER_JARS: נתיבי Cloud Storage של מנהלי התקנים של JDBC, מופרדים באמצעות פסיקים
• ‫DRIVER_CLASS_NAME: שם המחלקה של מנהל ההתקן של JDBC
• ‫CONNECTION_URL: מחרוזת כתובת ה-URL של חיבור JDBC.
• ‫OUTPUT_TABLE: טבלת הפלט ב-BigQuery
• ‫BIG_QUERY_LOADING_TEMPORARY_DIRECTORY: ספריית הזמנית לתהליך הטעינה של BigQuery

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",
     "parameters": {
       "driverJars": "DRIVER_JARS",
       "driverClassName": "DRIVER_CLASS_NAME",
       "connectionURL": "CONNECTION_URL",
       "outputTable": "OUTPUT_TABLE",
       "bigQueryLoadingTemporaryDirectory": "BIG_QUERY_LOADING_TEMPORARY_DIRECTORY",
     },
     "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Jdbc_to_BigQuery_Flex",
     "environment": { "maxWorkers": "10" }
  }
}

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

• ‫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
• ‫DRIVER_JARS: נתיבי Cloud Storage של מנהלי התקנים של JDBC, מופרדים באמצעות פסיקים
• ‫DRIVER_CLASS_NAME: שם המחלקה של מנהל ההתקן של JDBC
• ‫CONNECTION_URL: מחרוזת כתובת ה-URL של חיבור JDBC.
• ‫OUTPUT_TABLE: טבלת הפלט ב-BigQuery
• ‫BIG_QUERY_LOADING_TEMPORARY_DIRECTORY: ספריית הזמנית לתהליך הטעינה של BigQuery

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

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 סיפקה