Spark结构化流+Kafka集成:MicroBatchExecution PartitionOffsets错误

Question

我使用Spark Structured Streaming分别处理从Apache Kafka传入和传出的数据流，使用下面的scala代码。

我可以使用kafka源成功读取数据流，但在尝试将流写入kafka接收器时，我遇到了以下错误：

ERROR MicroBatchExecution:91 - Query [id = 234750ca-d416-4182-b3cc-4e2c1f922724, runId = 4c4b0931-9876-456f-8d56-752623803332] terminated with error java.lang.IllegalArgumentException: Expected e.g. {"topicA":{"0":23,"1":-1},"topicB":{"0":-2}}, got 1 {"path":"file:///path/to/file.csv","timestamp":1536564701000,"batchId":0}
at org.apache.spark.sql.kafka010.JsonUtils$.partitionOffsets(JsonUtils.scala:74)
at org.apache.spark.sql.kafka010.KafkaSourceOffset$.apply(KafkaSourceOffset.scala:64)
at org.apache.spark.sql.kafka010.KafkaSource$$anon$1.deserialize(KafkaSource.scala:124)
at org.apache.spark.sql.kafka010.KafkaSource$$anon$1.deserialize(KafkaSource.scala:99)
at org.apache.spark.sql.execution.streaming.HDFSMetadataLog.get(HDFSMetadataLog.scala:198)
at org.apache.spark.sql.kafka010.KafkaSource.initialPartitionOffsets$lzycompute(KafkaSource.scala:129)
at org.apache.spark.sql.kafka010.KafkaSource.initialPartitionOffsets(KafkaSource.scala:97)
at org.apache.spark.sql.kafka010.KafkaSource.getBatch(KafkaSource.scala:207)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$populateStartOffsets$2.apply(MicroBatchExecution.scala:216)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$org$apache$spark$sql$execution$streaming$MicroBatchExecution$$populateStartOffsets$2.apply(MicroBatchExecution.scala:213)
at scala.collection.Iterator$class.foreach(Iterator.scala:893)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
at org.apache.spark.sql.execution.streaming.StreamProgress.foreach(StreamProgress.scala:25)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.org$apache$spark$sql$execution$streaming$MicroBatchExecution$$populateStartOffsets(MicroBatchExecution.scala:213)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply$mcV$sp(MicroBatchExecution.scala:124)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:121)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1$$anonfun$apply$mcZ$sp$1.apply(MicroBatchExecution.scala:121)
at org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:271)
at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:58)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$runActivatedStream$1.apply$mcZ$sp(MicroBatchExecution.scala:121)
at org.apache.spark.sql.execution.streaming.ProcessingTimeExecutor.execute(TriggerExecutor.scala:56)
at org.apache.spark.sql.execution.streaming.MicroBatchExecution.runActivatedStream(MicroBatchExecution.scala:117)
at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:279)
at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:189)
Exception in thread "main" org.apache.spark.sql.streaming.StreamingQueryException: Expected e.g. {"topicA":{"0":23,"1":-1},"topicB":{"0":-2}}, got 1
{"path":""file:///path/to/file.csv"","timestamp":1536564701000,"batchId":0}
=== Streaming Query ===
Identifier: [id = 234750ca-d416-4182-b3cc-4e2c1f922724, runId = 851d0cd7-aabe-45c8-8a14-94227f90e174]
Current Committed Offsets: {KafkaSource[Subscribe[t]]: {"logOffset":2}}
Current Available Offsets: {KafkaSource[Subscribe[t]]: {"logOffset":3}}

Scala代码：

object spark_kafka_attempt2 {
def main(args: Array[String]) {
val spark = SparkSession
.builder
.appName("spark_kafka_test")
.getOrCreate()
import spark.implicits._
val input_lines = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094")
.option("subscribe", "input_stream")
.option("startingOffsets", "earliest")
.load()
val inputStreamSchema = new StructType()
.add("input_id", "long")
.add("timestamp", "timestamp")
.add("type", "string")
val lines = input_lines.selectExpr("CAST(value AS STRING)", "CAST(timestamp AS TIMESTAMP)").as[(String, Timestamp)]
.select(from_json($"value", inputStreamSchema).as("data"), $"timestamp".as("arrival_timestamp"))
.select("data.*", "arrival_timestamp")

val query = lines
.selectExpr("CAST(input_id AS STRING) AS key", "to_json(struct(*)) AS value")
.writeStream
.format("kafka")
.outputMode("update")
.option("kafka.bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094")
.option("topic", "processed_stream")
.option("checkpointLocation", "/home/local/directory")
.start()
query.awaitTermination()
}
}

当输出被发送到控制台时，代码运行良好，而当试图将处理后的流发送到Apache Kafka时出现错误。

我使用的是Apache结构化流媒体2.3.1、Scala 2.11.8和Apache Kafka 2.0。

Build.sbt文件如下：

name := "spark_kafka_test"    
version := "0.1"    
scalaVersion := "2.11.8"    
val sparkVersion = "2.3.1"    
libraryDependencies ++= Seq(
"org.apache.spark" %% "spark-sql" % sparkVersion,
"org.apache.spark" %% "spark-sql-kafka-0-10" % sparkVersion
) 

我提交我的工作火花如下：

./spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.1 --class spark_kafka_test --master local[4] /home/salman/Development/spark_kafka_attempt2/target/scala-2.11/spark_kafka_test_2.11-0.1.jar 

Answer 1

经过大量的调查和浏览，我找到了以下将处理后的流写入kafka接收器的解决方案：

创建以下KafkaSink类

import java.util.Properties
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}    
import org.apache.spark.sql.ForeachWriter
class  KafkaSink(topic:String, servers:String) extends ForeachWriter[(String, String)]
{
val kafkaProperties = new Properties()
kafkaProperties.put("bootstrap.servers", servers)
kafkaProperties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
kafkaProperties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
val results = new scala.collection.mutable.HashMap[String, String]
var producer: KafkaProducer[String, String] = _
def open(partitionId: Long,version: Long): Boolean = {
producer = new KafkaProducer(kafkaProperties)
true
}
def process(value: (String, String)): Unit = {
producer.send(new ProducerRecord(topic, value._1 + ":" + value._2))
}
def close(errorOrNull: Throwable): Unit = {
producer.close()
}
}

使用Foreach编写器将数据发送到kafkasink，如下所示：

val outputDf = lines.selectExpr("CAST(input_id AS STRING) AS key", "to_json(struct(*)) AS value").as[(String, String)]
val topic = "processed_stream"
val brokers = "localhost:9092,localhost:9093,localhost:9094"
val writer = new KafkaSink(topic, brokers)
val query = outputDf
.writeStream
.foreach(writer)
.outputMode("update")
.start()