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

בטא

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

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

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

• המינוי ל-Pub/Sub צריך להתקיים לפני שמריצים את צינור הנתונים.
• מקור ה-JDBC צריך להתקיים לפני שמריצים את צינור הנתונים.
• נושא ה-dead-letter של הפלט ב-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. עוברים לדף Create job from template (יצירת משימה מתבנית) ב-Dataflow.
כניסה לדף Create job from template
2. בשדה שם המשימה, מזינים שם ייחודי למשימה.
3. אופציונלי: בשדה Regional endpoint (נקודת קצה אזורית), בוחרים ערך מהתפריט הנפתח. אזור ברירת המחדל הוא us-central1.

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

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

gcloud

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

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

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location gs://dataflow-templates-REGION_NAME/VERSION/flex/ \
    --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 סיפקה.