Spark Parallel Stream - 对象不可序列化

Question

我正在使用 Spark 的多输入流 reader 来读取来自 Kafka 的消息。我收到下面提到的错误。如果我不使用多输入流 reader ，我不会收到任何错误。为了达到性能，我需要使用并行概念，测试目的我只使用一个。

错误

java.io.NotSerializableException: org.apache.kafka.clients.consumer.ConsumerRecord
Serialization stack:
    - object not serializable (class: org.apache.kafka.clients.consumer.ConsumerRecord, value: ConsumerRecord(topic = test, partition = 0, offset = 120, CreateTime = -1, checksum = 2372777361, serialized key size = -1, serialized value size = 48, key = null, value = 10051,2018-03-15 17:12:24+0000,Bentonville,Gnana))
    at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
    at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
    at org.apache.spark.serializer.SerializationStream.writeValue(Serializer.scala:134)
    at org.apache.spark.storage.DiskBlockObjectWriter.write(DiskBlockObjectWriter.scala:239)
    at org.apache.spark.shuffle.sort.BypassMergeSortShuffleWriter.write(BypassMergeSortShuffleWriter.java:151)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
    at org.apache.spark.scheduler.Task.run(Task.scala:108)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
18/03/15 17:12:24 ERROR TaskSetManager: Task 0.0 in stage 470.0 (TID 470) had a not serializable result: org.apache.kafka.clients.consumer.ConsumerRecord

代码：

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.Success
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Milliseconds, StreamingContext}

object ParallelStreamJob {

  def main(args: Array[String]): Unit = {
    val spark = SparkHelper.getOrCreateSparkSession()
    val ssc = new StreamingContext(spark.sparkContext, Milliseconds(50))
    val kafkaStream = {

      val kafkaParams = Map[String, Object](
        "bootstrap.servers" -> "localhost:9092",
        "key.deserializer" -> classOf[StringDeserializer],
        "value.deserializer" -> classOf[StringDeserializer],
        "group.id" -> "use_a_separate_group_id_for_each_stream",
        "auto.offset.reset" -> "latest",
        "enable.auto.commit" -> (false: java.lang.Boolean)
      )

      val topics = Array("test")
      val numPartitionsOfInputTopic = 1
      val streams = (1 to numPartitionsOfInputTopic) map { _ =>
        KafkaUtils.createDirectStream[String, String](
          ssc,
          PreferConsistent,
          Subscribe[String, String](topics, kafkaParams)
        )
      }
      val unifiedStream = ssc.union(streams)
      val sparkProcessingParallelism = 1 
      unifiedStream.repartition(sparkProcessingParallelism)
    }

    kafkaStream.foreachRDD(rdd=> {
      rdd.foreach(conRec=> {
        println(conRec.value())
      })
    })

    println(" Spark parallel reader is ready !!!")

   ssc.start()
    ssc.awaitTermination()

  }
}

sbt

scalaVersion := "2.11.8"
val sparkVersion = "2.2.0"
val connectorVersion = "2.0.7"
val kafka_stream_version = "1.6.3"

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % sparkVersion ,
  "org.apache.spark" %% "spark-sql" % sparkVersion  ,
  "org.apache.spark" %% "spark-hive" % sparkVersion  ,
  "com.datastax.spark" %% "spark-cassandra-connector" % connectorVersion  ,
  "org.apache.kafka" %% "kafka" % "0.10.1.0",
  "org.apache.spark" %% "spark-streaming-kafka-0-10" % sparkVersion,
  "org.apache.spark" %% "spark-streaming" %  sparkVersion  ,
)

如何解决这个问题？

Answer 1

问题很清楚java.io.NotSerializableException:org.apache.kafka.clients.consumer.ConsumerRecord。 ConsumerRecord class 不扩展 Serializable

尝试在foreachRdd操作kafkaStream.map(_.value())之前取出ConsumerRecord的value字段。

更新 1：上述修复不起作用，因为异常发生在 ssc.union(streams)。ssc.union(streams) 需要节点之间的数据传输，它必须序列化数据。所以，你可以在union操作之前通过map去掉value字段来解决这个问题。

KafkaUtils.createDirectStream[String, String]( ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParam) ).map(_.value())

Answer 2

首先，如果你有 1 个主题，那么你不应该使用创建多个 Kafkastreams，因为你使用的是直接方法，它会自动创建与有多个 Kafka 分区一样多的线程 topic.Spark 如果您遵循 DirectApproach，将自动负责并行化您的任务。尝试在 RDD 级别使用 repartition() 而不是重新分区 Dstream 本身。

Spark Parallel Stream - 对象不可序列化

Spark Parallel Stream - object not serializable

scala

apache-spark

spark-streaming