大佬教程收集整理的这篇文章主要介绍了Azure DataBricks Stream foreach失败,并显示NotSerializableException,大佬教程大佬觉得挺不错的,现在分享给大家,也给大家做个参考。
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
我想不断详细说明数据集流的行(最初由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
谁能解释为什么发生此异常以及如何避免?谢谢!
以上是大佬教程为你收集整理的Azure DataBricks Stream foreach失败,并显示NotSerializableException全部内容,希望文章能够帮你解决Azure DataBricks Stream foreach失败,并显示NotSerializableException所遇到的程序开发问题。
如果觉得大佬教程网站内容还不错,欢迎将大佬教程推荐给程序员好友。
本图文内容来源于网友网络收集整理提供,作为学习参考使用,版权属于原作者。
如您有任何意见或建议可联系处理。小编QQ:384754419,请注明来意。