DataFrame, возвращаемый getBatch из MQTTTextStreamSource, не имеет isStreaming=true

Question

Я пытаюсь использовать MQTT вместе с PySpark Structured Streaming.

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode
from pyspark.sql.functions import split

spark = SparkSession \
    .builder \
    .appName("Test") \
    .master("local[4]") \
    .getOrCreate()

# Custom Structured Streaming receiver
lines = spark\
             .readStream\
             .format("org.apache.bahir.sql.streaming.mqtt.MQTTStreamSourceProvider")\
             .option("topic","uwb/distances")\
             .option('brokerUrl', 'tcp://127.0.0.1:1883')\
             .load()

# Split the lines into words
words = lines.select(explode(split(lines.value, ' ')).alias('word'))

# Generate running word count
wordCounts = words.groupBy('word').count()

# Start running the query that prints the running counts to the console
query = wordCounts \
    .writeStream \
    .outputMode('complete') \
    .format('console') \
    .start()

query.awaitTermination()

Сообщение об ошибке:

Logical Plan:
Aggregate [word#7], [word#7, count(1) AS count#11L]
+- Project [word#7]
   +- Generate explode(split(value#2,  )), false, [word#7]
      +- StreamingExecutionRelation org.apache.bahir.sql.streaming.mqtt.MQTTTextStreamSource@383ccec1, [value#2, timestamp#3]

    at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:295)
    at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:189)
Caused by: java.lang.AssertionError: assertion failed: DataFrame returned by getBatch from org.apache.bahir.sql.streaming.mqtt.MQTTTextStreamSource@383ccec1 did not have isStreaming=true

Я не понимаю, что не так в моем коде. Более того, согласно этому посту, структурированная потоковая передача 2.1.0 фактически поддерживается Bahir MQTT. Я также попробовал Spark 2.2.1 и получил ту же проблему.

Вот как я запускаю код:

spark-submit \
  --jars lib/spark-streaming-mqtt_2.11-2.2.1.jar, \
  lib/spark-sql-streaming-mqtt_2.11-2.2.1.jar, \
  lib/org.eclipse.paho.client.mqttv3-1.2.0.jar \
  TestSpark.py

Как я могу решить эту проблему?

Answer 1

Я скачал двоичные файлы Spark 2.2.0 и выполнил код следующим образом:

~/Downloads/spark-2.2.1-bin-hadoop2.7/bin/spark-submit \
    --jars lib/spark-streaming-mqtt_2.11-2.2.1.jar, \
    lib/spark-sql-streaming-mqtt_2.11-2.2.1.jar, \
    lib/org.eclipse.paho.client.mqttv3-1.2.0.jar \
    TestSpark.py

Это решило проблему. Ранее я только менял версии файлов jar MQTT, например, spark-streaming-mqtt_2.11-2.2.1.jar, но, видимо, этого было недостаточно.