scala - Neo4j Spark Streaming Scala 的 NotSerializableException
问题描述
我正在尝试使用 Neo4j-Spark 连接器在 Neo4j 中运行查询。我想将流中的值(由 Kafka 作为字符串生成)传递到我的查询中。但是,我得到序列化异常:
Caused by: java.io.NotSerializableException: org.apache.spark.SparkContext
Serialization stack:
- object not serializable (class: org.apache.spark.SparkContext, value: org.apache.spark.SparkContext@54688d9f)
- field (class: consumer.SparkConsumer$$anonfun$processingLogic$2, name: sc$1, type: class org.apache.spark.SparkContext)
- object (class consumer.SparkConsumer$$anonfun$processingLogic$2, <function1>)
- field (class: consumer.SparkConsumer$$anonfun$processingLogic$2$$anonfun$apply$3, name: $outer, type: class consumer.SparkConsumer$$anonfun$processingLogic$2)
- object (class consumer.SparkConsumer$$anonfun$processingLogic$2$$anonfun$apply$3, <function1>)
下面是主要功能和查询逻辑的代码:
object SparkConsumer {
def main(args: Array[String]) {
val config = "neo4j_local"
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("KafkaSparkStreaming")
setNeo4jSparkConfig(config, sparkConf)
val sparkSession = SparkSession
.builder()
.config(sparkConf)
.getOrCreate()
val streamingContext = new StreamingContext(sparkSession.sparkContext, Seconds(3))
streamingContext.sparkContext.setLogLevel("ERROR")
val sqlContext = new SQLContext(streamingContext.sparkContext)
val numStreams = 2
val topics = Array("member_topic1")
def kafkaParams(i: Int) = Map[String, Object](
"bootstrap.servers" -> "localhost:9092",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"group.id" -> "group2",
"auto.offset.reset" -> "latest",
"enable.auto.commit" -> (false: java.lang.Boolean)
)
val lines = (1 to numStreams).map(i => KafkaUtils.createDirectStream[String, String](
streamingContext,
LocationStrategies.PreferConsistent,
ConsumerStrategies.Subscribe[String, String](topics, kafkaParams(i))
))
val messages = streamingContext.union(lines)
val wordsArrays = values.map(_.split(","))
wordsArrays.foreachRDD(rdd => rdd.foreach(
data => execNeo4jSearchQuery(data)(streamingContext.sparkContext)
))
streamingContext.start()
streamingContext.awaitTermination()
}
def execNeo4jSearchQuery(data: Array[String])(implicit sc: SparkContext) = {
val neo = Neo4j(sc)
val query = "my query"
val paramsMap = Map("lat" -> data(1).toDouble, "lon" -> data(2).toDouble, "id" -> data(0).toInt)
val df = neo.cypher(query, paramsMap).loadDataFrame("group_name" -> "string", "event_name" -> "string", "venue_name" -> "string", "distance" -> "double")
println("\ndf:")
df.show()
}
}
解决方案
不允许从 executor 访问或创建分布式数据结构SparkContext
。SparkSession
所以:
wordsArrays.foreachRDD(rdd => rdd.foreach(
data => execNeo4jSearchQuery(data)(streamingContext.sparkContext)
))
在哪里execNeo4jSearchQuery
调用:
neo.cypher(query, paramsMap).loadDataFrame
不是有效的 Spark 代码。
如果你想直接访问 Neo4j,RDD.foreach
你必须使用标准客户端(AnormCypher似乎提供了非常优雅的 API),而不需要转换为 Spark 分布式结构。
有点不相关的说明-您可能会考虑对记录集使用单个连接foreachPartition
(也SPARK Cost of Initalizing Database Connection in map / mapPartitions context)。
推荐阅读
- docker - 从 mknod 创建的容器中删除循环设备
- java - 使用 jstack 清空 Tomcat 线程转储
- go - 如何在 Go 中加载 Angular 构建文件
- javascript - 如何使用 javascript 在 wp 中执行现有的短代码
- corda - Corda 流量更改不适用于 Corda OS 4.0
- javascript - 返回 Angular 反应形式的选定属性内的所有对象值
- powershell - 基于变量的投射
- c# - UWP MediaCapture 初始化失败
- ssh - 如何在不使用临时外部 IP 的情况下使用 Pycharm 与 GCP VM 进行 SSH?
- key - ETCD 键名 - 支持哪些字符?