תבנית JMS ל-Pub/Sub

בטא

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

תבנית ה-JMS to Pub/Sub היא צינור עיבוד נתונים בסטרימינג שקורא הודעות מ-Active MQ JMS Server (תור/נושא) וכותב אותן ל-Pub/Sub.

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

• השם של נושא הפלט ב-Pub/Sub חייב להיות קיים.
• כתובת ה-IP של מארח ה-JMS צריכה להתקיים ולהיות לה הגדרת רשת מתאימה כדי שמכונות וירטואליות של עובדי Dataflow יוכלו להגיע למארח ה-JMS.
• לנושא או לתור של JMS שממנו מופקים הנתונים צריך להיות שם.

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

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

• ‫inputName: השם של נושא או תור JMS שממנו נתונים נקראים. לדוגמה, queue.
• ‫inputType: סוג היעד של JMS שממנו יקראו נתונים. יכול להיות תור או נושא. לדוגמה, queue.
• ‫outputTopic: השם של נושא Pub/Sub שאליו יתפרסמו הנתונים. לדוגמה, projects/<PROJECT_ID>/topics/<TOPIC_NAME>.
• ‫username: שם המשתמש שמשמש לאימות בשרת JMS. לדוגמה, sampleusername.
• ‫password: הסיסמה שמשויכת לשם המשתמש שסופק. לדוגמה, samplepassword.

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

• ‫jmsServer: כתובת ה-IP של שרת ה-JMS ‏ (ActiveMQ). לדוגמה, tcp://10.0.0.1:61616.

הרצת התבנית

המסוף

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

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

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

gcloud

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

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

gcloud dataflow flex-template run JOB_NAME \
    --project=YOUR_PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/JMS_to_Cloud_PubSub \
    --parameters \
jmsServer=JMS_SERVER,\
inputName=INPUT_NAME,\
inputType=INPUT_TYPE,\
outputTopic=OUTPUT_TOPIC,\
username=USERNAME,\
password=PASSWORD
  

בדוגמה הזו צריך להחליף את הערכים הבאים:

• מחליפים את YOUR_PROJECT_ID במזהה הפרויקט.
• מחליפים את בשם האזור של Dataflow. לדוגמה: us-central1.
• מחליפים את JOB_NAME בשם של המשימה לבחירתכם. שם המשרה חייב להתאים לביטוי הרגולרי [a-z]([-a-z0-9]{0,38}[a-z0-9])? כדי להיחשב כתקין.
• מחליפים את JMS_SERVER בכתובות של שרת ה-JMS. לדוגמה: tcp://10.0.0.0:61616
• מחליפים את INPUT_NAME בשם של נושא או תור הקלט של שרת JMS. לדוגמה: testtopic.
• מחליפים את INPUT_TYPE ב-Destination Type (סוג היעד) של שרת ה-JMS (תור או נושא). לדוגמה: topic
• מחליפים את OUTPUT_TOPIC בשם של נושא הפלט ב-Pub/Sub. לדוגמה: projects/myproject/topics/testoutput.
• מחליפים את USERNAME בשם המשתמש של שרת ה-JMS. לדוגמה: testuser.
• מחליפים את PASSWORD בסיסמה שמתאימה לשם המשתמש שמשמש את שרת ה-JMS.

API

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

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "jmsServer": "JMS_SERVER",
          "inputName": "INPUT_NAME",
          "inputType": "INPUT_TYPE",
          "outputTopic": "OUTPUT_TOPIC",
          "username": "USERNAME",
          "password": "PASSWORD"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-REGION_NAME/VERSION/flex/JMS_to_Cloud_PubSub",
   }
}
  

בדוגמה הזו צריך להחליף את הערכים הבאים:

• מחליפים את YOUR_PROJECT_ID במזהה הפרויקט.
• מחליפים את בשם האזור של Dataflow. לדוגמה: us-central1.
• מחליפים את JOB_NAME בשם של המשימה לבחירתכם. שם המשרה חייב להתאים לביטוי הרגולרי [a-z]([-a-z0-9]{0,38}[a-z0-9])? כדי להיחשב כתקין.
• מחליפים את JMS_SERVER בכתובות של שרת ה-JMS. לדוגמה: tcp://10.0.0.0:61616
• מחליפים את INPUT_NAME בשם של נושא או תור הקלט של שרת JMS. לדוגמה: testtopic.
• מחליפים את INPUT_TYPE ב-Destination Type (סוג היעד) של שרת ה-JMS (תור או נושא). לדוגמה: topic
• מחליפים את OUTPUT_TOPIC בשם של נושא הפלט ב-Pub/Sub. לדוגמה: projects/myproject/topics/testoutput.
• מחליפים את USERNAME בשם המשתמש של שרת ה-JMS. לדוגמה: testuser.
• מחליפים את PASSWORD בסיסמה שמתאימה לשם המשתמש שמשמש את שרת ה-JMS.

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

Java

/*
 * Copyright (C) 2023 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 com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.metadata.TemplateParameter;
import javax.jms.ConnectionFactory;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
import org.apache.beam.sdk.io.jms.JmsIO;
import org.apache.beam.sdk.io.jms.JmsRecord;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.PCollection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Dataflow template which reads data from Jms Queue/Topic and writes it to Cloud PubSub.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/jms-to-pubsub/README_Jms_to_PubSub.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Jms_to_PubSub",
    category = TemplateCategory.STREAMING,
    displayName = "JMS to Pubsub",
    description =
        "The JMS to Pub/Sub template is a streaming pipeline that reads messages from ActiveMQ JMS Server (Queue/Topic) and writes them to Pub/Sub.",
    optionsClass = com.google.cloud.teleport.v2.templates.JmsToPubsub.JmsToPubsubOptions.class,
    flexContainerName = "googlecloud-and-jms",
    contactInformation = "https://cloud.google.com/support",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/jms-to-pubsub",
    preview = true,
    requirements = {
      "The Pub/Sub output topic name must exist.",
      "The JMS host IP must exist and have the proper network configuration for Dataflow worker VMs to reach the JMS host.",
      "The JMS topic/queue that data is extracted from must have a name."
    },
    streaming = true,
    supportsAtLeastOnce = true)
public class JmsToPubsub {

  /**
   * Runs a pipeline which reads data from JMS queue/topic and writes it to Cloud PubSub.
   *
   * @param args arguments to the pipeline
   */
  private static final Logger LOG = LoggerFactory.getLogger(JmsToPubsub.class);

  public static void main(String[] args) {
    JmsToPubsubOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(JmsToPubsubOptions.class);
    run(options);
  }

  public static void validate(JmsToPubsubOptions options) {
    if (options != null) {
      if ((options.getUsername() != null
              && (!options.getUsername().isEmpty() || !options.getUsername().isBlank()))
          && (options.getPassword() == null
              || options.getPassword().isBlank()
              || options.getPassword().isEmpty())) {
        throw new IllegalArgumentException(
            "While username is provided, password is required for authentication");
      }
    }
  }

  public static PipelineResult run(JmsToPubsubOptions options) {
    validate(options);
    Pipeline pipeline = Pipeline.create(options);
    String connectionURI = options.getJmsServer();
    ConnectionFactory myConnectionFactory;
    PCollection<JmsRecord> input;

    if (!options.getUsername().isEmpty() || !options.getUsername().isBlank()) {
      myConnectionFactory =
          new ActiveMQConnectionFactory(
              options.getUsername(), options.getPassword(), connectionURI);
    } else {
      myConnectionFactory = new ActiveMQConnectionFactory(connectionURI);
    }
    LOG.info("Given Input Type " + options.getInputType());
    if (options.getInputType().equalsIgnoreCase("queue")) {
      input =
          pipeline.apply(
              "Read From JMS Queue",
              JmsIO.read()
                  .withConnectionFactory(myConnectionFactory)
                  .withQueue(options.getInputName()));
    } else {
      input =
          pipeline.apply(
              "Read From JMS Topic",
              JmsIO.read()
                  .withConnectionFactory(myConnectionFactory)
                  .withTopic(options.getInputName()));
    }

    input
        .apply(
            MapElements.via(
                new SimpleFunction<JmsRecord, String>() {
                  public String apply(JmsRecord input) {
                    return input.getPayload();
                  }
                }))
        .apply("WriteToPubSubTopic", PubsubIO.writeStrings().to(options.getOutputTopic()));
    return pipeline.run();
  }

  /**
   * The {@link JmsToPubsubOptions} interface provides the custom execution options passed by the
   * executor at the command-line.
   */
  public interface JmsToPubsubOptions extends PipelineOptions {
    @TemplateParameter.Text(
        order = 1,
        groupName = "Source",
        optional = true,
        regexes = {"[,\\/:a-zA-Z0-9._-]+"},
        description = "JMS Host IP",
        helpText = "The JMS (ActiveMQ) Server IP.",
        example = "tcp://10.0.0.1:61616")
    @Validation.Required
    String getJmsServer();

    void setJmsServer(String jmsServer);

    @TemplateParameter.Text(
        order = 2,
        groupName = "Source",
        optional = false,
        regexes = {"[a-zA-Z0-9._-]+"},
        description = "JMS Queue/Topic Name to read the input from",
        helpText = "The name of the JMS topic or queue that data is read from.",
        example = "queue")
    @Validation.Required
    String getInputName();

    void setInputName(String inputName);

    @TemplateParameter.Text(
        order = 3,
        optional = false,
        regexes = {"[a-zA-Z0-9._-]+"},
        description = "JMS Destination Type to read the input from",
        helpText = "The JMS destination type to read data from. Can be a queue or a topic.",
        example = "queue")
    @Validation.Required
    String getInputType();

    void setInputType(String inputType);

    @TemplateParameter.PubsubTopic(
        order = 4,
        groupName = "Target",
        description = "Output Pub/Sub topic",
        helpText = "The name of the Pub/Sub topic to publish data to.",
        example = "projects/<PROJECT_ID>/topics/<TOPIC_NAME>")
    @Validation.Required
    String getOutputTopic();

    void setOutputTopic(String outputTopic);

    @TemplateParameter.Text(
        order = 5,
        description = "JMS Username",
        helpText = "The username to use for authentication on the JMS server.",
        example = "sampleusername")
    String getUsername();

    void setUsername(String username);

    @TemplateParameter.Password(
        order = 6,
        description = "JMS Password",
        helpText = "The password associated with the provided username.",
        example = "samplepassword")
    String getPassword();

    void setPassword(String password);
  }
}

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

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