在 Apache Beam 中使用 Protobuf 编码的 MQTT 流答案

【问题标题】：Working with Protobuf-encoded MQTT streams in Apache Beam在 Apache Beam 中使用 Protobuf 编码的 MQTT 流
【发布时间】：2021-01-18 18:53:30
【问题描述】：

我正在尝试使用 Apache Beam 解码和处理 protobuf 编码的 MQTT 消息（来自 Eclipse Mosquitto 代理）。除了编码字段，我还想处理每条消息的完整主题以进行分组和聚合，以及时间戳。

到目前为止我所尝试的

我可以通过

连接到 Mosquitto

val options = PipelineOptionsFactory.create()
val pipeline = Pipeline.create(options)

val mqttReader: MqttIO.Read = MqttIO
    .read()
    .withConnectionConfiguration(
        MqttIO.ConnectionConfiguration.create(
            "tcp://localhost:1884",
            "my/topic/+"
        )
    )

val readMessages = pipeline.apply<PCollection<ByteArray>>(mqttReader)

为了解码消息，我通过 Gradle 编译了 .proto 架构（在我的情况下为 quote.proto，其中包含 Quote 消息），这允许我通过 @ 将 ByteArray 转换为 Quote 对象987654330@:

val quotes = readMessages
    .apply(
        ParDo.of(object : DoFn<ByteArray, QuoteOuterClass.Quote>() {
            @ProcessElement
            fun processElement(context: ProcessContext) {
                val protoRow = context.element()
                context.output(QuoteOuterClass.Quote.parseFrom(protoRow))
            }
        })
    )

使用它，在下一个 apply 中，我可以使用 ProcessFunction 和 lambda 访问各个字段，例如{ quote -> "${quote.volume}" }。但是，有两个问题：

使用此管道，我无法访问每条消息的主题或时间戳。
在使用纯 UTF8 编码将解码后的消息发送回代理后，我认为它们没有被正确解码。

其他注意事项

Apache Beam 提供了ProtoCoder class，但我不知道如何将它与 MqttIO 结合使用。我怀疑实现必须看起来类似于

    val coder = ProtoCoder
        .of(QuoteOuterClass.Quote::class.java)
        .withExtensionsFrom(QuoteOuterClass::class.java)

Kafka IO 阅读器提供的不是PCollection<ByteArray>，而是PCollection<KafkaRecord<Long, String>>，它包含所有相关字段（包括主题）。我想知道 Mqtt + ProtoBuf 是否可以实现类似的功能。
可以在 Spark Structured Streaming + Apache Bahir 中完成与我想要实现的类似的实现，如下所示：

val df_mqttStream = spark.readStream
    .format("org.apache.bahir.sql.streaming.mqtt.MQTTStreamSourceProvider")
    .option("topic", topic)
    .load(brokerUrl)

val parsePayload = ProtoSQL.udf { bytes: Array[Byte] => Quote.parseFrom(bytes) }

val quotesDS = df_mqttStream.select("id", "topic", "payload")
    .withColumn("quote", parsePayload($"payload"))
    .select("id", "topic", "quote.*")

但是，在 Spark 2.4（支持的最新版本）中，访问消息主题会中断（related issue、my ticket in Apache Jira）。

【问题讨论】：

标签： apache-spark protocol-buffers mqtt apache-beam

【解决方案1】：

据我了解，最新版本的 Apache Beam (2.27.0) 根本没有提供提取 MQTT 消息特定主题的方法。

我已扩展 MqttIO 以返回包含 topic（和 timestamp）以及字节数组 payload 的 MqttMessage 对象。这些更改目前以pull request draft 的形式存在。

通过这些更改，可以通过message.topic 轻松访问该主题。

val readMessages = pipeline.apply<PCollection<MqttMessage>>(mqttReader)

val topicOfMessages: PCollection<String> = mqttMessages
    .apply(
        ParDo.of(object : DoFn<MqttMessage, String>() {
            @ProcessElement
            fun processElement(
                @Element message: MqttMessage,
                out: OutputReceiver<String>
            ) { out.output(message.topic) }
        })
    )

【讨论】：