תבנית MQTT ל-Pub/Sub

בטא

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

תבנית MQTT ל-Pub/Sub היא צינור עיבוד נתונים בסטרימינג שקורא הודעות מנושא MQTT וכותב אותן ל-Pub/Sub. הוא כולל את הפרמטרים האופציונליים username ו-password למקרה ששרת ה-MQTT דורש אימות.

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

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

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

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

• ‫inputTopic: השם של נושא ה-MQTT שממנו הנתונים נקראים. לדוגמה, topic.
• ‫outputTopic: השם של נושא הפלט ב-Pub/Sub שהנתונים נכתבים אליו. לדוגמה, projects/your-project-id/topics/your-topic-name.
• ‫username: שם המשתמש שמשמש לאימות בשרת MQTT. לדוגמה, sampleusername.
• ‫password: הסיסמה שמשויכת לשם המשתמש שסופק. לדוגמה, samplepassword.

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

• ‫brokerServer: כתובת ה-IP או המארח של שרת מתווך MQTT. לדוגמה, tcp://host:1883.

הרצת התבנית

המסוף

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

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

4. בתפריט הנפתח Dataflow template (תבנית של Dataflow), בוחרים באפשרות MQTT to Pub/Sub template (תבנית MQTT ל-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/VERSION/flex/Mqtt_to_PubSub \
    --parameters \
brokerServer=MQTT_SERVER,\
inputTopic=INPUT_TOPIC,\
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])? כדי להיחשב כתקין.
• מחליפים את INPUT_TOPIC בשם של נושא הקלט של שרת MQTT. לדוגמה: testtopic.
• מחליפים את MQTT_SERVER בכתובות של שרת ה-MQTT. לדוגמה: tcp://10.128.0.62:1883
• מחליפים את OUTPUT_TOPIC בשם של נושא הפלט ב-Pub/Sub. לדוגמה: projects/myproject/topics/testoutput.
• מחליפים את USERNAME בשם המשתמש של שרת ה-MQTT. לדוגמה: testuser.
• מחליפים את PASSWORD בסיסמה שמתאימה לשם המשתמש שמשמש עם שרת ה-MQTT.

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": {
          "brokerServer": "MQTT_SERVER",
          "inputTopic": "INPUT_TOPIC",
          "outputTopic": "OUTPUT_TOPIC",
          "username": "USERNAME",
          "password": "PASSWORD"
      },
      "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/Mqtt_to_PubSub",
   }
}
  

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

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

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

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 java.nio.charset.StandardCharsets;
import java.util.regex.Pattern;
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.mqtt.MqttIO;
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.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Dataflow template which reads data from Mqtt Topic and writes it to Cloud PubSub.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/mqtt-to-pubsub/README_Mqtt_to_PubSub.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "Mqtt_to_PubSub",
    category = TemplateCategory.STREAMING,
    displayName = "MQTT to Pubsub",
    description =
        "The MQTT to Pub/Sub template is a streaming pipeline that reads messages from an MQTT topic and writes them to Pub/Sub. "
            + "It includes the optional parameters <code>username</code> and <code>password</code> in case authentication is required by the MQTT server.",
    optionsClass = MqttToPubsub.MqttToPubsubOptions.class,
    flexContainerName = "mqtt-to-pubsub",
    contactInformation = "https://cloud.google.com/support",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/mqtt-to-pubsub",
    preview = true,
    requirements = {
      "The Pub/Sub output topic name must exist.",
      "The MQTT host IP must exist and have the proper network configuration for worker machines to reach the MQTT host.",
      "The MQTT topic that data is extracted from must have a name."
    },
    streaming = true,
    supportsAtLeastOnce = true)
public class MqttToPubsub {

  private static final Logger LOG = LoggerFactory.getLogger(MqttToPubsub.class);
  private static final Pattern PUBSUB_TOPIC_PATTERN =
      Pattern.compile("^projects/[\\w:-]+/topics/[\\w:-]+$");

  /**
   * Runs a pipeline which reads data from Mqtt topic and writes it to Cloud PubSub.
   *
   * @param args arguments to the pipeline
   */
  public static void main(String[] args) {
    MqttToPubsubOptions options =
        PipelineOptionsFactory.fromArgs(args).withValidation().as(MqttToPubsubOptions.class);
    run(options);
  }

  public static void validate(MqttToPubsubOptions options) {
    if (options != null) {
      if ((options.getUsername() != null && !options.getUsername().isEmpty())
          && (options.getPassword() == null || options.getPassword().isBlank())) {
        throw new IllegalArgumentException(
            "While username is provided, password is required for authentication");
      }
      if (options.getInputTopic() != null
          && PUBSUB_TOPIC_PATTERN.matcher(options.getInputTopic()).matches()) {
        LOG.warn(
            "The input topic '{}' matches the format of a Google Cloud Pub/Sub topic. "
                + "Please verify that this is a valid MQTT topic and not a misconfigured Pub/Sub topic.",
            options.getInputTopic());
      }
    }
  }

  public static PipelineResult run(MqttToPubsubOptions options) {
    validate(options);
    Pipeline pipeline = Pipeline.create(options);
    MqttIO.Read<byte[]> mqttIo;
    if (!options.getUsername().isEmpty() || !options.getPassword().isBlank()) {
      mqttIo =
          MqttIO.read()
              .withConnectionConfiguration(
                  MqttIO.ConnectionConfiguration.create(
                          options.getBrokerServer(), options.getInputTopic())
                      .withUsername(options.getUsername())
                      .withPassword(options.getPassword()));
    } else {
      mqttIo =
          MqttIO.read()
              .withConnectionConfiguration(
                  MqttIO.ConnectionConfiguration.create(
                      options.getBrokerServer(), options.getInputTopic()));
    }

    return pipeline
        .apply("ReadFromMqttTopic", mqttIo)
        .apply(ParDo.of(new ByteToStringTransform()))
        .apply("WriteToPubSubTopic", PubsubIO.writeStrings().to(options.getOutputTopic()))
        .getPipeline()
        .run();
  }

  static class ByteToStringTransform extends DoFn<byte[], String> {
    @ProcessElement
    public void processElement(@Element byte[] word, OutputReceiver<String> out) {
      out.output(new String(word, StandardCharsets.UTF_8));
    }
  }

  /**
   * The {@link MqttToPubsubOptions} interface provides the custom execution options passed by the
   * executor at the command-line.
   */
  public interface MqttToPubsubOptions extends PipelineOptions {
    @TemplateParameter.Text(
        order = 1,
        groupName = "Source",
        optional = true,
        regexes = {"[,\\/:a-zA-Z0-9._-]+"},
        description = "MQTT Broker IP",
        helpText = "The MQTT broker server IP or host.",
        example = "tcp://host:1883")
    @Validation.Required
    String getBrokerServer();

    void setBrokerServer(String brokerServer);

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

    void setInputTopic(String inputTopics);

    @TemplateParameter.PubsubTopic(
        order = 3,
        groupName = "Target",
        description = "Output Pub/Sub topic",
        helpText = "The name of the output Pub/Sub topic that data is written to.",
        example = "projects/your-project-id/topics/your-topic-name")
    @Validation.Required
    String getOutputTopic();

    void setOutputTopic(String outputTopic);

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

    void setUsername(String username);

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

    void setPassword(String password);
  }
}

בעיות מוכרות

לפני Apache Beam 2.76.0, אם צינור הנתונים לא מקבל הודעה מנושא MQTT במשך יותר מ-90 דקות, מתרחשת StackOverflowError. כדי לפתור את הבעיה, צריך לעדכן את התבנית לגרסה העדכנית ביותר.

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

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