在 spark 版本 2.2.0 中使用 python(pyspark) 从 mqtt 获取数据流

Posted 2023-04-15

【中文标题】在 spark 版本 2.2.0 中使用 python(pyspark) 从 mqtt 获取数据流

【英文标题】：Get stream of data from mqtt using python(pyspark) in spark version 2.2.0

【发布时间】：2017-09-06 17:05:07

【问题描述】：

我在 spark 的 POM.xml 中添加了依赖项，在以下链接中给出：

http://bahir.apache.org/docs/spark/current/spark-sql-streaming-mqtt/

再次使用 maven 构建 spark。但正如我们所见，它只显示了 Java 和 Scala 支持从 mqtt 获取数据。

我想从 python 中的 mqtt 获取流数据。 在早期版本中，我们有一个相同的 pyspark.streaming.mqtt。 spark 2.2.0 pyspark 中有什么相似之处。 我正在使用 mosquitto 作为 mqtt 代理。

【参考方案1】：

对于 PySpark，您可以使用 Structured Streaming 绑定（您必须包含 Bahir jar）：

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()  # type: SparkSession
(spark
    .readStream
    .format("org.apache.bahir.sql.streaming.mqtt.MQTTStreamSourceProvider")
    .load("tcp://".format(broker_uri)))