KryoException：无法通过 spark 结构化流找到 class
KryoException: Unable to find class with spark structured streaming
1-问题
我有一个 Spark 程序，它使用 Kryo 但不是 [=66= 的一部分]火花力学。更具体地说，我正在使用 Spark Structured Streaming 连接到 Kafka.
I read binary values coming from Kafka and decode it on my own.
我在尝试使用 Kryo 反序列化数据时遇到异常。然而，只有当我打包我的程序并将其 运行 放在 Spark 独立集群 上时，才会发生此问题。也就是说，当我 运行 它时，它不会发生，在 intellij 中，即 Spark 本地模式（开发模式）。
我得到的异常如下：
Caused by: com.esotericsoftware.kryo.KryoException: Unable to find
class: com.elsevier.entellect.commons.package$RawData
请注意，RawData 是我自己的一个案例 Class，位于我的多项目构建的子项目之一中。
要了解上下文，请在下面找到更多详细信息：
2-build.sbt:
lazy val commonSettings = Seq(
organization := "com.elsevier.entellect",
version := "0.1.0-SNAPSHOT",
scalaVersion := "2.11.12",
resolvers += Resolver.mavenLocal,
updateOptions := updateOptions.value.withLatestSnapshots(false)
)
lazy val entellectextractors = (project in file("."))
.settings(commonSettings).aggregate(entellectextractorscommon, entellectextractorsfetchers, entellectextractorsmappers, entellectextractorsconsumers)
lazy val entellectextractorscommon = project
.settings(
commonSettings,
libraryDependencies ++= Seq(
"com.esotericsoftware" % "kryo" % "5.0.0-RC1",
"com.github.romix.akka" %% "akka-kryo-serialization" % "0.5.0" excludeAll(excludeJpountz),
"org.apache.kafka" % "kafka-clients" % "1.0.1",
"com.typesafe.akka" %% "akka-stream" % "2.5.16",
"com.typesafe.akka" %% "akka-http-spray-json" % "10.1.4",
"com.typesafe.akka" % "akka-slf4j_2.11" % "2.5.16",
"ch.qos.logback" % "logback-classic" % "1.2.3"
)
)
lazy val entellectextractorsfetchers = project
.settings(
commonSettings,
libraryDependencies ++= Seq(
"com.typesafe.akka" %% "akka-stream-kafka" % "0.22",
"com.typesafe.slick" %% "slick" % "3.2.3",
"com.typesafe.slick" %% "slick-hikaricp" % "3.2.3",
"com.lightbend.akka" %% "akka-stream-alpakka-slick" % "0.20")
)
.dependsOn(entellectextractorscommon)
lazy val entellectextractorsconsumers = project
.settings(
commonSettings,
libraryDependencies ++= Seq(
"com.typesafe.akka" %% "akka-stream-kafka" % "0.22")
)
.dependsOn(entellectextractorscommon)
lazy val entellectextractorsmappers = project
.settings(
commonSettings,
mainClass in assembly := Some("entellect.extractors.mappers.NormalizedDataMapper"),
assemblyMergeStrategy in assembly := {
case PathList("META-INF", "services", "org.apache.spark.sql.sources.DataSourceRegister") => MergeStrategy.concat
case PathList("META-INF", xs @ _*) => MergeStrategy.discard
case x => MergeStrategy.first},
dependencyOverrides += "com.fasterxml.jackson.core" % "jackson-core" % "2.9.5",
dependencyOverrides += "com.fasterxml.jackson.core" % "jackson-databind" % "2.9.5",
dependencyOverrides += "com.fasterxml.jackson.module" % "jackson-module-scala_2.11" % "2.9.5",
dependencyOverrides += "org.apache.jena" % "apache-jena" % "3.8.0",
libraryDependencies ++= Seq(
"org.apache.jena" % "apache-jena" % "3.8.0",
"edu.isi" % "karma-offline" % "0.0.1-SNAPSHOT",
"org.apache.spark" % "spark-core_2.11" % "2.3.1" % "provided",
"org.apache.spark" % "spark-sql_2.11" % "2.3.1" % "provided",
"org.apache.spark" %% "spark-sql-kafka-0-10" % "2.3.1"
//"com.datastax.cassandra" % "cassandra-driver-core" % "3.5.1"
))
.dependsOn(entellectextractorscommon)
lazy val excludeJpountz = ExclusionRule(organization = "net.jpountz.lz4", name = "lz4")
包含spark代码的子项目是entellectextractorsmappers。包含无法找到的案例class RawData 的子项目是entellectextractorscommon。 entellectextractorsmappers 明确依赖于 entellectextractorscommon.
3- 在本地独立集群上提交和在本地开发模式下运行提交的区别：
当我提交到集群时，我的 spark 依赖项如下：
"org.apache.spark" % "spark-core_2.11" % "2.3.1" % "provided",
"org.apache.spark" % "spark-sql_2.11" % "2.3.1" % "provided",
当我 运行 在本地开发模式下（没有提交脚本）他们变成这样
"org.apache.spark" % "spark-core_2.11" % "2.3.1",
"org.apache.spark" % "spark-sql_2.11" % "2.3.1",
也就是说，在本地开发中我需要有依赖项，而在以独立模式提交到集群时，它们已经在集群中，所以我按提供的方式放置它们。
4-如何提交:
spark-submit --class entellect.extractors.mappers.DeNormalizedDataMapper --name DeNormalizedDataMapper --master spark://MaatPro.local:7077 --deploy-mode cluster --executor-memory 14G --num-executors 1 --conf spark.sql.shuffle.partitions=7 "/Users/maatari/IdeaProjects/EntellectExtractors/entellectextractorsmappers/target/scala-2.11/entellectextractorsmappers-assembly-0.1.0-SNAPSHOT.jar"
5-我如何使用 Kryo:
5.1-申报登记
在 entellectextractorscommon 项目中，我有一个包含以下内容的包对象：
package object commons {
case class RawData(modelName: String,
modelFile: String,
sourceType: String,
deNormalizedVal: String,
normalVal: Map[String, String])
object KryoContext {
lazy val kryoPool = new Pool[Kryo](true, false, 16) {
protected def create(): Kryo = {
val kryo = new Kryo()
kryo.setRegistrationRequired(false)
kryo.addDefaultSerializer(classOf[scala.collection.Map[_,_]], classOf[ScalaImmutableAbstractMapSerializer])
kryo.addDefaultSerializer(classOf[scala.collection.generic.MapFactory[scala.collection.Map]], classOf[ScalaImmutableAbstractMapSerializer])
kryo.addDefaultSerializer(classOf[RawData], classOf[ScalaProductSerializer])
kryo
}
}
lazy val outputPool = new Pool[Output](true, false, 16) {
protected def create: Output = new Output(4096)
}
lazy val inputPool = new Pool[Input](true, false, 16) {
protected def create: Input = new Input(4096)
}
}
object ExecutionContext {
implicit lazy val system = ActorSystem()
implicit lazy val mat = ActorMaterializer()
implicit lazy val ec = system.dispatcher
}
}
5.2-用法
在 entellectextractorsmappers（spark 程序所在的位置）中，我使用 mapMartition。在其中，我有一种方法来解码来自 kafka 的数据，该方法使用 Kryo 本身：
def decodeData(rowOfBinaryList: List[Row], kryoPool: Pool[Kryo], inputPool: Pool[Input]): List[RawData] = {
val kryo = kryoPool.obtain()
val input = inputPool.obtain()
val data = rowOfBinaryList.map(r => r.getAs[Array[Byte]]("message")).map{ binaryMsg =>
input.setInputStream(new ByteArrayInputStream(binaryMsg))
val value = kryo.readClassAndObject(input).asInstanceOf[RawData]
input.close()
value
}
kryoPool.free(kryo)
inputPool.free(input)
data
}
注意：对象 KryoContext + Lazy val 确保每个 JVM 实例化一次 kryoPool。不过，我认为问题不在于此。
I red in some other place a hint about issues of classLoaders used by
spark vs Kryo? But not sure to really understand what is going on.
如果有人能给我一些指点，那会有所帮助，因为我不知道从哪里开始。为什么它会在本地模式而不是集群模式下工作，所提供的是否会混淆依赖关系并在 Kryo 中造成一些问题？是 SBT Assembly 合并策略搞砸了吗？
可能有很多建议，如果有人可以帮助我缩小范围，那就太棒了！
到目前为止，
我已经通过选择 "enclosing" class 加载器解决了这个问题，我想它是来自 Spark 的加载器。这是在准备好一些关于 Kryo 和 Spark 之间的 Class Loader 问题的评论之后：
lazy val kryoPool = new Pool[Kryo](true, false, 16) {
protected def create(): Kryo = {
val cl = Thread.currentThread().getContextClassLoader()
val kryo = new Kryo()
kryo.setClassLoader(cl)
kryo.setRegistrationRequired(false)
kryo.addDefaultSerializer(classOf[scala.collection.Map[_,_]], classOf[ScalaImmutableAbstractMapSerializer])
kryo.addDefaultSerializer(classOf[scala.collection.generic.MapFactory[scala.collection.Map]], classOf[ScalaImmutableAbstractMapSerializer])
kryo.addDefaultSerializer(classOf[RawData], classOf[ScalaProductSerializer])
kryo
}
}
1-问题
我有一个 Spark 程序，它使用 Kryo 但不是 [=66= 的一部分]火花力学。更具体地说，我正在使用 Spark Structured Streaming 连接到 Kafka.
I read binary values coming from Kafka and decode it on my own.
我在尝试使用 Kryo 反序列化数据时遇到异常。然而，只有当我打包我的程序并将其 运行 放在 Spark 独立集群 上时，才会发生此问题。也就是说，当我 运行 它时，它不会发生，在 intellij 中，即 Spark 本地模式（开发模式）。
我得到的异常如下：
Caused by: com.esotericsoftware.kryo.KryoException: Unable to find class: com.elsevier.entellect.commons.package$RawData
请注意，RawData 是我自己的一个案例 Class，位于我的多项目构建的子项目之一中。
要了解上下文，请在下面找到更多详细信息：
2-build.sbt:
lazy val commonSettings = Seq(
organization := "com.elsevier.entellect",
version := "0.1.0-SNAPSHOT",
scalaVersion := "2.11.12",
resolvers += Resolver.mavenLocal,
updateOptions := updateOptions.value.withLatestSnapshots(false)
)
lazy val entellectextractors = (project in file("."))
.settings(commonSettings).aggregate(entellectextractorscommon, entellectextractorsfetchers, entellectextractorsmappers, entellectextractorsconsumers)
lazy val entellectextractorscommon = project
.settings(
commonSettings,
libraryDependencies ++= Seq(
"com.esotericsoftware" % "kryo" % "5.0.0-RC1",
"com.github.romix.akka" %% "akka-kryo-serialization" % "0.5.0" excludeAll(excludeJpountz),
"org.apache.kafka" % "kafka-clients" % "1.0.1",
"com.typesafe.akka" %% "akka-stream" % "2.5.16",
"com.typesafe.akka" %% "akka-http-spray-json" % "10.1.4",
"com.typesafe.akka" % "akka-slf4j_2.11" % "2.5.16",
"ch.qos.logback" % "logback-classic" % "1.2.3"
)
)
lazy val entellectextractorsfetchers = project
.settings(
commonSettings,
libraryDependencies ++= Seq(
"com.typesafe.akka" %% "akka-stream-kafka" % "0.22",
"com.typesafe.slick" %% "slick" % "3.2.3",
"com.typesafe.slick" %% "slick-hikaricp" % "3.2.3",
"com.lightbend.akka" %% "akka-stream-alpakka-slick" % "0.20")
)
.dependsOn(entellectextractorscommon)
lazy val entellectextractorsconsumers = project
.settings(
commonSettings,
libraryDependencies ++= Seq(
"com.typesafe.akka" %% "akka-stream-kafka" % "0.22")
)
.dependsOn(entellectextractorscommon)
lazy val entellectextractorsmappers = project
.settings(
commonSettings,
mainClass in assembly := Some("entellect.extractors.mappers.NormalizedDataMapper"),
assemblyMergeStrategy in assembly := {
case PathList("META-INF", "services", "org.apache.spark.sql.sources.DataSourceRegister") => MergeStrategy.concat
case PathList("META-INF", xs @ _*) => MergeStrategy.discard
case x => MergeStrategy.first},
dependencyOverrides += "com.fasterxml.jackson.core" % "jackson-core" % "2.9.5",
dependencyOverrides += "com.fasterxml.jackson.core" % "jackson-databind" % "2.9.5",
dependencyOverrides += "com.fasterxml.jackson.module" % "jackson-module-scala_2.11" % "2.9.5",
dependencyOverrides += "org.apache.jena" % "apache-jena" % "3.8.0",
libraryDependencies ++= Seq(
"org.apache.jena" % "apache-jena" % "3.8.0",
"edu.isi" % "karma-offline" % "0.0.1-SNAPSHOT",
"org.apache.spark" % "spark-core_2.11" % "2.3.1" % "provided",
"org.apache.spark" % "spark-sql_2.11" % "2.3.1" % "provided",
"org.apache.spark" %% "spark-sql-kafka-0-10" % "2.3.1"
//"com.datastax.cassandra" % "cassandra-driver-core" % "3.5.1"
))
.dependsOn(entellectextractorscommon)
lazy val excludeJpountz = ExclusionRule(organization = "net.jpountz.lz4", name = "lz4")
包含spark代码的子项目是entellectextractorsmappers。包含无法找到的案例class RawData 的子项目是entellectextractorscommon。 entellectextractorsmappers 明确依赖于 entellectextractorscommon.
3- 在本地独立集群上提交和在本地开发模式下运行提交的区别：
当我提交到集群时，我的 spark 依赖项如下：
"org.apache.spark" % "spark-core_2.11" % "2.3.1" % "provided",
"org.apache.spark" % "spark-sql_2.11" % "2.3.1" % "provided",
当我 运行 在本地开发模式下（没有提交脚本）他们变成这样
"org.apache.spark" % "spark-core_2.11" % "2.3.1",
"org.apache.spark" % "spark-sql_2.11" % "2.3.1",
也就是说，在本地开发中我需要有依赖项，而在以独立模式提交到集群时，它们已经在集群中，所以我按提供的方式放置它们。
4-如何提交:
spark-submit --class entellect.extractors.mappers.DeNormalizedDataMapper --name DeNormalizedDataMapper --master spark://MaatPro.local:7077 --deploy-mode cluster --executor-memory 14G --num-executors 1 --conf spark.sql.shuffle.partitions=7 "/Users/maatari/IdeaProjects/EntellectExtractors/entellectextractorsmappers/target/scala-2.11/entellectextractorsmappers-assembly-0.1.0-SNAPSHOT.jar"
5-我如何使用 Kryo:
5.1-申报登记
在 entellectextractorscommon 项目中，我有一个包含以下内容的包对象：
package object commons {
case class RawData(modelName: String,
modelFile: String,
sourceType: String,
deNormalizedVal: String,
normalVal: Map[String, String])
object KryoContext {
lazy val kryoPool = new Pool[Kryo](true, false, 16) {
protected def create(): Kryo = {
val kryo = new Kryo()
kryo.setRegistrationRequired(false)
kryo.addDefaultSerializer(classOf[scala.collection.Map[_,_]], classOf[ScalaImmutableAbstractMapSerializer])
kryo.addDefaultSerializer(classOf[scala.collection.generic.MapFactory[scala.collection.Map]], classOf[ScalaImmutableAbstractMapSerializer])
kryo.addDefaultSerializer(classOf[RawData], classOf[ScalaProductSerializer])
kryo
}
}
lazy val outputPool = new Pool[Output](true, false, 16) {
protected def create: Output = new Output(4096)
}
lazy val inputPool = new Pool[Input](true, false, 16) {
protected def create: Input = new Input(4096)
}
}
object ExecutionContext {
implicit lazy val system = ActorSystem()
implicit lazy val mat = ActorMaterializer()
implicit lazy val ec = system.dispatcher
}
}
5.2-用法
在 entellectextractorsmappers（spark 程序所在的位置）中，我使用 mapMartition。在其中，我有一种方法来解码来自 kafka 的数据，该方法使用 Kryo 本身：
def decodeData(rowOfBinaryList: List[Row], kryoPool: Pool[Kryo], inputPool: Pool[Input]): List[RawData] = {
val kryo = kryoPool.obtain()
val input = inputPool.obtain()
val data = rowOfBinaryList.map(r => r.getAs[Array[Byte]]("message")).map{ binaryMsg =>
input.setInputStream(new ByteArrayInputStream(binaryMsg))
val value = kryo.readClassAndObject(input).asInstanceOf[RawData]
input.close()
value
}
kryoPool.free(kryo)
inputPool.free(input)
data
}
注意：对象 KryoContext + Lazy val 确保每个 JVM 实例化一次 kryoPool。不过，我认为问题不在于此。
I red in some other place a hint about issues of classLoaders used by spark vs Kryo? But not sure to really understand what is going on.
如果有人能给我一些指点，那会有所帮助，因为我不知道从哪里开始。为什么它会在本地模式而不是集群模式下工作，所提供的是否会混淆依赖关系并在 Kryo 中造成一些问题？是 SBT Assembly 合并策略搞砸了吗？
可能有很多建议，如果有人可以帮助我缩小范围，那就太棒了！
到目前为止，
我已经通过选择 "enclosing" class 加载器解决了这个问题，我想它是来自 Spark 的加载器。这是在准备好一些关于 Kryo 和 Spark 之间的 Class Loader 问题的评论之后：
lazy val kryoPool = new Pool[Kryo](true, false, 16) {
protected def create(): Kryo = {
val cl = Thread.currentThread().getContextClassLoader()
val kryo = new Kryo()
kryo.setClassLoader(cl)
kryo.setRegistrationRequired(false)
kryo.addDefaultSerializer(classOf[scala.collection.Map[_,_]], classOf[ScalaImmutableAbstractMapSerializer])
kryo.addDefaultSerializer(classOf[scala.collection.generic.MapFactory[scala.collection.Map]], classOf[ScalaImmutableAbstractMapSerializer])
kryo.addDefaultSerializer(classOf[RawData], classOf[ScalaProductSerializer])
kryo
}
}