我想继续详细说明数据集流的行(最初由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
谁能解释为什么会发生这种异常以及如何避免?谢谢!
这个问题与以下两个有关:
Spark Context不可序列化。
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