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

בטא

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

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

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

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

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

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

• ‫driverClassName: השם של מחלקת מנהל ההתקן של JDBC. לדוגמה, com.mysql.jdbc.Driver.
• ‫connectionUrl: מחרוזת כתובת ה-URL של חיבור JDBC. אפשר להעביר את הערך הזה כמחרוזת שמוצפנת באמצעות מפתח Cloud KMS ואז מקודדת ב-Base64. לדוגמה: 'echo -n "jdbc:mysql://some-host:3306/sampledb" | gcloud kms encrypt --location=
• ‫driverJars: נתיבי Cloud Storage מופרדים בפסיקים לדרייברים של JDBC. לדוגמה, gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar.
• ‫query: השאילתה להרצה במקור כדי לחלץ את הנתונים. לדוגמה, select * from sampledb.sample_table.
• ‫outputTopic: נושא Pub/Sub לפרסום. לדוגמה, projects/<PROJECT_ID>/topics/<TOPIC_NAME>.

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

• ‫username: שם המשתמש שמשמש לחיבור JDBC. אפשר להעביר את הערך הזה כמחרוזת שמוצפנת באמצעות מפתח Cloud KMS ואז מקודדת ב-Base64. לדוגמה, echo -n 'some_username' | glcloud kms encrypt --location=my_location --keyring=mykeyring --key=mykey --plaintext-file=- --ciphertext-file=- | base64.
• ‫password: הסיסמה לשימוש בחיבור JDBC. אפשר להעביר את הערך הזה כמחרוזת שמוצפנת באמצעות מפתח Cloud KMS ואז מקודדת ב-Base64. לדוגמה, echo -n 'some_password' | glcloud kms encrypt --location=my_location --keyring=mykeyring --key=mykey --plaintext-file=- --ciphertext-file=- | base64.
• ‫connectionProperties: מחרוזת המאפיינים לשימוש בחיבור JDBC. הפורמט של המחרוזת צריך להיות [propertyName=property;]*. לדוגמה, unicode=true;characterEncoding=UTF-8.
• ‫KMSEncryptionKey: מפתח ההצפנה של Cloud KMS שבו יש להשתמש כדי לפענח את שם המשתמש, הסיסמה ומחרוזת החיבור. אם מועבר מפתח Cloud KMS, צריך להעביר את שם המשתמש, הסיסמה ומחרוזת החיבור כשהם מוצפנים ומקודדים ב-Base64. לדוגמה, projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key.
• ‫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>.

הרצת התבנית

המסוף

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

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

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

gcloud

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

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

gcloud dataflow jobs run JOB_NAME \
    --gcs-location gs://dataflow-templates/VERSION/flex/Jdbc_to_PubSub \
    --region REGION_NAME \
    --parameters \
driverClassName=DRIVER_CLASS_NAME,\
connectionURL=JDBC_CONNECTION_URL,\
driverJars=DRIVER_PATHS,\
username=CONNECTION_USERNAME,\
password=CONNECTION_PASSWORD,\
connectionProperties=CONNECTION_PROPERTIES,\
query=SOURCE_SQL_QUERY,\
outputTopic=OUTPUT_TOPIC,\
KMSEncryptionKey=KMS_ENCRYPTION_KEY

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

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

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

  • latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית. אפשר למצוא את הגרסה הזו בתיקיית ההורה המתאימה עם התאריך בדלי – gs://dataflow-templates/
  זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרות בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫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, אם יש צורך
• ‫SOURCE_SQL_QUERY: שאילתת ה-SQL שרוצים להריץ במסד הנתונים של המקור
• ‫OUTPUT_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
{
  "launchParameter": {
    "jobName": "JOB_NAME",
    "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/Jdbc_to_PubSub"
    "parameters": {
      "driverClassName": "DRIVER_CLASS_NAME",
      "connectionURL": "JDBC_CONNECTION_URL",
      "driverJars": "DRIVER_PATHS",
      "username": "CONNECTION_USERNAME",
      "password": "CONNECTION_PASSWORD",
      "connectionProperties": "CONNECTION_PROPERTIES",
      "query": "SOURCE_SQL_QUERY",
      "outputTopic": "OUTPUT_TOPIC",
      "KMSEncryptionKey":"KMS_ENCRYPTION_KEY"
    },
    "environment": { "zone": "us-central1-f" }
  }
}

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

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

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

  • latest כדי להשתמש בגרסה העדכנית של התבנית, שזמינה בתיקיית ההורה ללא תאריך בדלי – gs://dataflow-templates/latest/
  • שם הגרסה, כמו 2023-09-12-00_RC00, כדי להשתמש בגרסה ספציפית של התבנית. אפשר למצוא את הגרסה הזו בתיקיית ההורה המתאימה עם התאריך בדלי – gs://dataflow-templates/
  זהירות: יכול להיות שעדכונים בגרסה העדכנית של התבניות יגרמו לשינויים שישברו את התאימות לאחור. כדי שהשינויים האלה לא ישפיעו על תהליכי העבודה שלכם בסביבת הייצור, בסביבות הייצור צריך להשתמש בתבניות שנשמרות בתיקייה הראשית עם תאריך העדכנית ביותר.
• ‫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, אם יש צורך
• ‫SOURCE_SQL_QUERY: שאילתת ה-SQL שרוצים להריץ במסד הנתונים של המקור
• ‫OUTPUT_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.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.options.JdbcToPubsubOptions;
import java.sql.Clob;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
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.StaticValueProvider;
import org.apache.beam.sdk.values.PCollection;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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

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

  /**
   * {@link JdbcIO.RowMapper} implementation to convert Jdbc ResultSet rows to UTF-8 encoded JSONs.
   */
  public static class ResultSetToJSONString implements JdbcIO.RowMapper<String> {

    @Override
    public String mapRow(ResultSet resultSet) throws Exception {
      ResultSetMetaData metaData = resultSet.getMetaData();
      JSONObject json = new JSONObject();

      for (int i = 1; i <= metaData.getColumnCount(); i++) {
        Object value = resultSet.getObject(i);

        // JSONObject.put() does not support null values. The exception is JSONObject.NULL
        if (value == null) {
          json.put(metaData.getColumnLabel(i), JSONObject.NULL);
          continue;
        }

        switch (metaData.getColumnTypeName(i).toLowerCase()) {
          case "clob":
            Clob clobObject = resultSet.getClob(i);
            if (clobObject.length() > Integer.MAX_VALUE) {
              LOG.warn(
                  "The Clob value size {} in column {} exceeds 2GB and will be truncated.",
                  clobObject.length(),
                  metaData.getColumnLabel(i));
            }
            json.put(
                metaData.getColumnLabel(i), clobObject.getSubString(1, (int) clobObject.length()));
            break;
          default:
            json.put(metaData.getColumnLabel(i), value);
        }
      }
      return json.toString();
    }
  }

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

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

    run(options);
  }

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

    LOG.info("Starting Jdbc-To-PubSub Pipeline.");

    /*
     * Steps:
     *  1) Read data from a Jdbc Table
     *  2) Write to Pub/Sub topic
     */
    JdbcIO.DataSourceConfiguration dataSourceConfiguration =
        JdbcIO.DataSourceConfiguration.create(
                StaticValueProvider.of(options.getDriverClassName()),
                maybeDecrypt(options.getConnectionUrl(), options.getKMSEncryptionKey()))
            .withDriverJars(options.getDriverJars());
    if (options.getUsername() != null) {
      dataSourceConfiguration =
          dataSourceConfiguration.withUsername(
              maybeDecrypt(options.getUsername(), options.getKMSEncryptionKey()));
    }
    if (options.getPassword() != null) {
      dataSourceConfiguration =
          dataSourceConfiguration.withPassword(
              maybeDecrypt(options.getPassword(), options.getKMSEncryptionKey()));
    }
    if (options.getConnectionProperties() != null) {
      dataSourceConfiguration =
          dataSourceConfiguration.withConnectionProperties(options.getConnectionProperties());
    }

    PCollection<String> jdbcData =
        pipeline.apply(
            "readFromJdbc",
            JdbcIO.<String>read()
                .withDataSourceConfiguration(dataSourceConfiguration)
                .withQuery(options.getQuery())
                .withCoder(StringUtf8Coder.of())
                .withRowMapper(new ResultSetToJSONString()));

    jdbcData.apply("writeSuccessMessages", PubsubIO.writeStrings().to(options.getOutputTopic()));

    return pipeline.run();
  }
}

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

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