Azure DataBricks Stream foreach失败，并显示NotSerializableException

Question

我想不断详细说明数据集流的行（最初由Kafka发起）：基于条件，我想更新Radis哈希。这是我的代码段（lastContacts是上一个命令的结果，该命令是这种类型的流：org.apache.spark.sql.DataFrame = [serialNumber: string, lastModified: long]。扩展为org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]）：

class MyStreamProcessor extends ForeachWriter[Row] {
  override def open(partitionId: Long, version: Long): Boolean = {
    true
  }

  override def process(record: Row) = {
    val stringHashRDD = sc.parallelize(Seq(("lastContact", record(1).toString)))
    sc.toRedisHASH(stringHashRDD, record(0).toString)(redisConfig)
  }

  override def close(errorOrNull: Throwable): Unit = {}
}

val query = lastContacts
  .writeStream
  .foreach(new MyStreamProcessor())
  .start()

query.awaitTermination()

我收到一个巨大的堆栈跟踪，相关部分（我认为）是这样的： java.io.NotSerializableException: org.apache.spark.sql.streaming.DataStreamWriter

谁能解释为什么发生此异常以及如何避免？谢谢！

Answer 1

Spark上下文不可序列化。

ForeachWriter的任何实现都必须可序列化，因为每个任务都将获得所提供对象的新的序列化反序列化副本。因此，强烈建议在调用open（…）方法之后执行任何用于写入数据的初始化操作（例如，打开连接或启动事务），这表明任务已准备好生成数据。

在您的代码中，您尝试在流程方法中使用spark上下文，

override def process(record: Row) = {
    val stringHashRDD = sc.parallelize(Seq(("lastContact", record(1).toString)))
    *sc.toRedisHASH(stringHashRDD, record(0).toString)(redisConfig)*
  }

要将数据发送到Redis，您需要创建自己的连接并以open方法打开它，然后在process方法中使用它。

看一下如何创建Redis连接池。https://github.com/RedisLabs/spark-
redis/blob/master/src/main/scala/com/redislabs/provider/redis/ConnectionPool.scala