Почему запрос выбросить ClassCastException "SerializedOffset не может быть приведен к org.apache.spark.sql.execution.streaming.LongOffset" с источником MQTT?
Я получаю следующее исключение, когда код Spark структурированного потокового
18/12/05 15:00:38 ОШИБКА StreamExecution: запрос [id = 48ec92a0-811a-4d57-a65d-c0b9c754e093, runId = 5e2adff4-855e-46c6-8592-05e3557544c6] завершен с ошибкой java.lang.ClassCastException.apache.spark.sql.execution.streaming.SerializedOffset не может быть приведен к org.apache.spark.sql.execution.streaming.LongOffset в org.apache.bahir.sql.streaming.mqtt.MQTTTextStreamSource.getBatch(MQTTStreamSource.ss.) в org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$ выполнение $ потоковое $StreamExecution$$runBatch$2$$anonfun$ применить $7.apply(StreamExecution.scala:614)
Это исключение происходит каждый раз, когда я запускаю запрос. Это работает, когда я запускаю его после удаления контрольной точки.
Структурированный потоковый код Spark, как показано ниже, в основном я просто читаю данные из очереди MQTT и записываю в индекс ElasticSearch.
spark
.readStream
.format("org.apache.bahir.sql.streaming.mqtt.MQTTStreamSourceProvider")
.option("topic", "Employee")
.option("username", "username")
.option("password", "password")
.option("clientId", "employee11")
.load("tcp://localhost:8000")
.as[(String, Timestamp)]
.writeStream
.outputMode("append")
.format("es")
.option("es.resource", "spark/employee")
.option("es.nodes", "localhost")
.option("es.port", 9200)
.start()
.awaitTermination()
Ниже приведены используемые зависимости. Я использую раздачу MapR.
"org.apache.spark" %% "spark-core" % "2.2.1-mapr-1803",
"org.apache.spark" %% "spark-sql" % "2.2.1-mapr-1803",
"org.apache.spark" %% "spark-streaming" % "2.2.1-mapr-1803",
"org.apache.bahir" %% "spark-sql-streaming-mqtt" % "2.2.1",
"org.apache.bahir" %% "spark-streaming-mqtt" % "2.2.1",
"org.elasticsearch" %% "elasticsearch-spark-20" % "6.3.2"
Команда Spark-submit
/opt/mapr/spark/spark-2.2.1/bin/spark-submit \
--master yarn \
--deploy-mode cluster \
--jars spark-sql-streaming-mqtt_2.11-2.2.1.jar,org.eclipse.paho.client.mqttv3-1.1.0.jar,elasticsearch-spark-20_2.11-6.3.2.jar,mail-1.4.7.jar myjar_2.11-0.1.jar \
--class <MAIN_CLASS>
Любая помощь в этом будет оценена.