תבנית Pub/Sub ל-Java Database Connectivity ‏ (JDBC)

בטא

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

תבנית Pub/Sub to Java Database Connectivity ‏ (JDBC) היא צינור עיבוד נתונים של סטרימינג שקולט נתונים ממנוי Pub/Sub קיים כמחרוזות JSON, וכותב את הרשומות שמתקבלות ל-JDBC.

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

• המינוי ל-Pub/Sub צריך להתקיים לפני שמריצים את צינור הנתונים.
• מקור ה-JDBC צריך להתקיים לפני שמריצים את צינור הנתונים.
• נושא להודעות ללא מוצא של הפלט ב-Pub/Sub צריך להתקיים לפני שמריצים את צינור הנתונים.

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

פרמטר	תיאור
driverClassName	שם המחלקה של מנהל ההתקן של JDBC. לדוגמה, com.mysql.jdbc.Driver.
connectionUrl	מחרוזת כתובת ה-URL של חיבור JDBC. לדוגמה, jdbc:mysql://some-host:3306/sampledb. אפשר להעביר את הערך הזה כמחרוזת שמוצפנת באמצעות מפתח Cloud KMS ואז מקודדת ב-Base64. מסירים את התווים של הרווחים מהמחרוזת שמקודדת ב-Base64.
driverJars	נתיבי Cloud Storage מופרדים בפסיקים למנהלי התקנים של JDBC. לדוגמה, gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar.
username	אופציונלי: שם המשתמש שישמש לחיבור JDBC. אפשר להעביר את הערך הזה מוצפן באמצעות מפתח Cloud KMS כמחרוזת בקידוד Base64.
password	אופציונלי: הסיסמה שתשמש לחיבור JDBC. אפשר להעביר את הערך הזה מוצפן באמצעות מפתח Cloud KMS כמחרוזת בקידוד Base64.
connectionProperties	אופציונלי: מחרוזת מאפיינים לשימוש בחיבור JDBC. הפורמט של המחרוזת צריך להיות [propertyName=property;]*. לדוגמה, unicode=true;characterEncoding=UTF-8.
statement	הצהרה להרצה מול מסד הנתונים. בהצהרה צריך לציין את שמות העמודות בטבלה, בכל סדר שהוא. רק הערכים של שמות העמודות שצוינו נקראים מ-JSON ומוסיפים אותם להצהרה. לדוגמה, INSERT INTO tableName (column1, column2) VALUES (?,?)
inputSubscription	מינוי הקלט של Pub/Sub שממנו יתבצע הקריאה, בפורמט projects/<project>/subscriptions/<subscription>.
outputDeadletterTopic	נושא Pub/Sub להעברת הודעות שלא ניתן למסור. לדוגמה, projects/<project-id>/topics/<topic-name>.
KMSEncryptionKey	אופציונלי: מפתח הצפנה של Cloud KMS לפענוח שם המשתמש, הסיסמה ומחרוזת החיבור. אם מועבר מפתח Cloud KMS, צריך להעביר את שם המשתמש, הסיסמה ומחרוזת החיבור כשהם מוצפנים.
extraFilesToStage	נתיבים ב-Cloud Storage או סודות ב-Secret Manager שמופרדים בפסיקים, לקבצים שרוצים להכין להעברה בעובד. הקבצים האלה יישמרו בספרייה /extra_files בכל עובד. לדוגמה, gs://<my-bucket>/file.txt,projects/<project-id>/secrets/<secret-id>/versions/<version-id>.

הפעלת התבנית

המסוף

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

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

4. בתפריט הנפתח Dataflow template (תבנית של העברת נתונים), בוחרים באפשרות the Pub/Sub to JDBC 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/Pubsub_to_Jdbc \
    --region REGION_NAME \
    --parameters \
driverClassName=DRIVER_CLASS_NAME,\
connectionUrl=JDBC_CONNECTION_URL,\
driverJars=DRIVER_PATHS,\
username=CONNECTION_USERNAME,\
password=CONNECTION_PASSWORD,\
connectionProperties=CONNECTION_PROPERTIES,\
statement=SQL_STATEMENT,\
inputSubscription=INPUT_SUBSCRIPTION,\
outputDeadletterTopic=OUTPUT_DEADLETTER_TOPIC,\
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
• ‫DRIVER_CLASS_NAME: שם המחלקה של ה-driver
• ‫JDBC_CONNECTION_URL: כתובת ה-URL של חיבור JDBC
• ‫DRIVER_PATHS: נתיבי Cloud Storage של מנהלי התקנים של JDBC, מופרדים באמצעות פסיקים
• ‫CONNECTION_USERNAME: שם המשתמש של חיבור ה-JDBC
• ‫CONNECTION_PASSWORD: הסיסמה לחיבור JDBC
• ‫CONNECTION_PROPERTIES: מאפייני החיבור של JDBC, אם יש צורך
• ‫SQL_STATEMENT: הצהרת ה-SQL שתופעל על מסד הנתונים
• ‫INPUT_SUBSCRIPTION: מינוי הקלט של Pub/Sub שממנו יתבצע קריאה
• ‫OUTPUT_DEADLETTER_TOPIC: Pub/Sub להעברת הודעות שלא ניתן למסור
• ‫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
{
   "jobName": "JOB_NAME",
   "parameters": {
       "driverClassName": "DRIVER_CLASS_NAME",
       "connectionUrl": "JDBC_CONNECTION_URL",
       "driverJars": "DRIVER_PATHS",
       "username": "CONNECTION_USERNAME",
       "password": "CONNECTION_PASSWORD",
       "connectionProperties": "CONNECTION_PROPERTIES",
       "statement": "SQL_STATEMENT",
       "inputSubscription": "INPUT_SUBSCRIPTION",
       "outputDeadletterTopic": "OUTPUT_DEADLETTER_TOPIC",
       "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
• ‫DRIVER_CLASS_NAME: שם המחלקה של ה-driver
• ‫JDBC_CONNECTION_URL: כתובת ה-URL של חיבור JDBC
• ‫DRIVER_PATHS: נתיבי Cloud Storage של מנהלי התקנים של JDBC, מופרדים באמצעות פסיקים
• ‫CONNECTION_USERNAME: שם המשתמש של חיבור ה-JDBC
• ‫CONNECTION_PASSWORD: הסיסמה לחיבור JDBC
• ‫CONNECTION_PROPERTIES: מאפייני החיבור של JDBC, אם יש צורך
• ‫SQL_STATEMENT: הצהרת ה-SQL שתופעל על מסד הנתונים
• ‫INPUT_SUBSCRIPTION: מינוי הקלט של Pub/Sub שממנו יתבצע קריאה
• ‫OUTPUT_DEADLETTER_TOPIC: Pub/Sub להעברת הודעות שלא ניתן למסור
• ‫KMS_ENCRYPTION_KEY: מפתח ההצפנה של Cloud KMS

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

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.utils.KMSUtils.maybeDecrypt;

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.io.DynamicJdbcIO;
import com.google.cloud.teleport.v2.options.PubsubToJdbcOptions;
import com.google.cloud.teleport.v2.transforms.ErrorConverters;
import com.google.cloud.teleport.v2.utils.JsonStringToQueryMapper;
import com.google.cloud.teleport.v2.values.FailsafeElement;
import com.google.common.base.Splitter;
import java.util.List;
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.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.io.jdbc.JdbcIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.values.PCollection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The {@link PubsubToJdbc} streaming pipeline reads data from Google Cloud PubSub and publishes to
 * JDBC.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/googlecloud-to-googlecloud/README_Pubsub_to_Jdbc.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Pubsub_to_Jdbc",
    category = TemplateCategory.STREAMING,
    displayName = "Pub/Sub to JDBC",
    description =
        "The Pub/Sub to Java Database Connectivity (JDBC) template is a streaming pipeline that ingests data from a "
            + "pre-existing Cloud Pub/Sub subscription as JSON strings, and writes the resulting records to JDBC.",
    optionsClass = PubsubToJdbcOptions.class,
    flexContainerName = "jdbc-and-googlecloud",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/pubsub-to-jdbc",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "The Cloud Pub/Sub subscription must exist prior to running the pipeline.",
      "The JDBC source must exist prior to running the pipeline.",
      "The Cloud Pub/Sub output deadletter topic must exist prior to running the pipeline.",
    },
    streaming = true,
    supportsAtLeastOnce = true)
public class PubsubToJdbc {

  /* Logger for class.*/
  private static final Logger LOG = LoggerFactory.getLogger(PubsubToJdbc.class);

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

  /**
   * Main entry point for pipeline execution.
   *
   * @param args Command line arguments to the pipeline.
   */
  public static void main(String[] args) {
    UncaughtExceptionLogger.register();

    PubsubToJdbcOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(PubsubToJdbcOptions.class);

    run(options);
  }

  /**
   * Runs a pipeline which reads message from Pub/Sub and writes to JDBC.
   *
   * @param options The execution options.
   * @return The pipeline result.
   */
  public static PipelineResult run(PubsubToJdbcOptions options) {
    // Create the pipeline
    Pipeline pipeline = Pipeline.create(options);

    LOG.info("Starting Pubsub-to-Jdbc Pipeline.");

    /*
     * Steps:
     *  1) Read data from a Pub/Sub subscription
     *  2) Write to Jdbc Table
     *  3) Write errors to deadletter topic
     */
    PCollection<String> pubsubData =
        pipeline.apply(
            "readFromPubSubSubscription",
            PubsubIO.readStrings().fromSubscription(options.getInputSubscription()));

    JdbcIO.DataSourceConfiguration dataSourceConfiguration =
        JdbcIO.DataSourceConfiguration.create(
                ValueProvider.StaticValueProvider.of(options.getDriverClassName()),
                maybeDecrypt(options.getConnectionUrl(), options.getKMSEncryptionKey()))
            .withDriverJars(options.getDriverJars());
    if (options.getUsername() != null) {
      dataSourceConfiguration =
          dataSourceConfiguration.withUsername(
              maybeDecrypt(options.getUsername(), options.getKMSEncryptionKey()).get());
    }
    if (options.getPassword() != null) {
      dataSourceConfiguration =
          dataSourceConfiguration.withPassword(
              maybeDecrypt(options.getPassword(), options.getKMSEncryptionKey()).get());
    }
    if (options.getConnectionProperties() != null) {
      dataSourceConfiguration =
          dataSourceConfiguration.withConnectionProperties(options.getConnectionProperties());
    }

    PCollection<FailsafeElement<String, String>> errors =
        pubsubData
            .apply(
                "writeToJdbc",
                DynamicJdbcIO.<String>write()
                    .withDataSourceConfiguration(dataSourceConfiguration)
                    .withStatement(options.getStatement())
                    .withPreparedStatementSetter(
                        new JsonStringToQueryMapper(getKeyOrder(options.getStatement()))))
            .setCoder(FAILSAFE_ELEMENT_CODER);

    errors.apply(
        "WriteFailedRecords",
        ErrorConverters.WriteStringMessageErrorsToPubSub.newBuilder()
            .setErrorRecordsTopic(options.getOutputDeadletterTopic())
            .build());

    return pipeline.run();
  }

  private static List<String> getKeyOrder(String statement) {
    int startIndex = statement.indexOf("(");
    int endIndex = statement.indexOf(")");
    String data = statement.substring(startIndex + 1, endIndex);
    return Splitter.on(',').trimResults().splitToList(data);
  }
}

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

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