EMR 生成的文件的 Spark Kryo 反序列化在本地失败
Spark Kryo deserialization of EMR-produced files fails locally
将 EMR 版本升级到 6.2.0（我们之前使用的是 5.0 beta - ish）和 Spark 3.0.1 后，我们注意到我们无法在本地读取从 EMR 集群写入的 Kryo 文件（这在以前显然是可能的） .尝试读取此类文件时，抛出的异常如下：
com.esotericsoftware.kryo.KryoException: java.lang.ClassCastException: scala.Tuple3 cannot be cast to scala.Tuple2
我们使用 spark 3.0.1 和 Kryo 4.0.2（捆绑）并使用 Kryo::readClassAndObject 读取 Kryo 文件，使用 SparkContext::sequenceFile.
在 RDD 上重新读取操作
长话短说：
AWS EMR 6.2.0（也可能更早）导致从 EMR 集群写入的 Kryo 文件的本地反序列化失败（由于集群 运行 正在使用 AWS Spark 分支）。 post.
末尾附有要修复的代码
最近，Amazon EMR 集群 运行 他们自己的 Apache Spark 分支（即，对于 EMR 6.2.0 集群，Spark 版本为 3.0.1.amzn-0），包括 Kryo作为我们自己使用的默认序列化框架。自从升级到 6.2.0 后，我们注意到我们无法在本地读取从 EMR 6.2.0 集群写入的 Kryo 文件，它们会失败并显示如下消息：
com.esotericsoftware.kryo.KryoException: java.lang.ClassCastException: scala.Tuple3 cannot be cast to scala.Tuple2
我们试图读取的 RDD 确实是 Tuple2 类型的 RDD，但显然在反序列化时，Kryo 出于某种原因认为它被编码为 Tuple3 的 RDD。
现在，在内部，Kryo 拥有一个 ID <-> class 的映射，该映射是在 运行 时构建的，预计在读取和写入 JVM 之间保持一致（用于找出class 反序列化到）。这个注册表建立在 Kryo 实例的实例化之上（我们使用 org.apache.spark.serializer.KryoSerializer::newKryo）。经过检查，我们注意到 Tuple2 的 ID 在执行序列化的 EMR 集群和我们的本地机器之间确实不同，并且差异归因于 EMR 设置中存在的单个 class 而不是本地 - 这个 class 是 org.apache.spark.scheduler.HighlyCompressedMapStatus$CompressedSizes ，它不存在于任何公开可用的 Spark 代码中，因此我们将其归因于 Amazon spark 分支。这实际上意味着我们无法在本地读取几乎任何由 EMR 集群写入的 class，因为不可能在本地使用 Spark 的分支，并且 class 在创建时注册在 ID 13 Kryo 实例（以后可能会明显改变），导致几乎所有 classes 反序列化失败。
这里丑陋的修复是使用 Kryo 实例的 ClassResolver。如果注册表中不存在 CompressedSizes class，我们将所有 ID x >= 13 的 classes 注册为 x + 1。这确实很丑陋，但作为本地修复，它可以工作。显然，EMR/Kryo/Spark的新版本也可能会中断，所以要格外小心（我们只在本地使用它进行调试，这仍然很多）。
代码：
以前，我们会像这样创建 Kryo 实例：
val kryoSerializer = new KryoSerializer(sc.getConf)
val kryo = kryoSerializer.newKryo()
现在，我们使用这个：
val kryo = adjustRegistrationsForEmrSpark(kryoSerializer.newKryo())
哪里
private def adjustRegistrationsForEmrSpark(kryo: Kryo): Kryo = {
val existingRegistrations = getRegistrations(kryo)
val emrSpecificClassExists = existingRegistrations.exists(_.getType.getName.contains("CompressedSizes"))
if (emrSpecificClassExists) {
println(s"detected emr-specific class when creating kryo, not making any adjustments")
kryo
} else {
println(s"emr-specific class missing from registrations, adjusting existing classes by an offset of 1 to compensate")
val classResolver = kryo.getClassResolver
existingRegistrations.filter(_.getId >= 13).foreach { registration =>
val toRegister = new Registration(registration.getType, registration.getSerializer, registration.getId + 1)
classResolver.register(toRegister)
}
kryo
}
}
private def getRegistrations(kryo: Kryo): List[Registration] = {
var classIndex = 0
var reg: Registration = null
var result: List[Registration] = List()
do {
reg = kryo.getClassResolver.getRegistration(classIndex)
if (reg != null) result ++= List(reg)
classIndex = classIndex + 1
} while (reg != null)
result
}
将 EMR 版本升级到 6.2.0（我们之前使用的是 5.0 beta - ish）和 Spark 3.0.1 后，我们注意到我们无法在本地读取从 EMR 集群写入的 Kryo 文件（这在以前显然是可能的） .尝试读取此类文件时，抛出的异常如下：
com.esotericsoftware.kryo.KryoException: java.lang.ClassCastException: scala.Tuple3 cannot be cast to scala.Tuple2
我们使用 spark 3.0.1 和 Kryo 4.0.2（捆绑）并使用 Kryo::readClassAndObject 读取 Kryo 文件，使用 SparkContext::sequenceFile.
在 RDD 上重新读取操作长话短说： AWS EMR 6.2.0（也可能更早）导致从 EMR 集群写入的 Kryo 文件的本地反序列化失败（由于集群 运行 正在使用 AWS Spark 分支）。 post.
末尾附有要修复的代码最近，Amazon EMR 集群 运行 他们自己的 Apache Spark 分支（即，对于 EMR 6.2.0 集群，Spark 版本为 3.0.1.amzn-0），包括 Kryo作为我们自己使用的默认序列化框架。自从升级到 6.2.0 后，我们注意到我们无法在本地读取从 EMR 6.2.0 集群写入的 Kryo 文件，它们会失败并显示如下消息：
com.esotericsoftware.kryo.KryoException: java.lang.ClassCastException: scala.Tuple3 cannot be cast to scala.Tuple2
我们试图读取的 RDD 确实是 Tuple2 类型的 RDD，但显然在反序列化时，Kryo 出于某种原因认为它被编码为 Tuple3 的 RDD。
现在，在内部，Kryo 拥有一个 ID <-> class 的映射，该映射是在 运行 时构建的，预计在读取和写入 JVM 之间保持一致（用于找出class 反序列化到）。这个注册表建立在 Kryo 实例的实例化之上（我们使用 org.apache.spark.serializer.KryoSerializer::newKryo）。经过检查，我们注意到 Tuple2 的 ID 在执行序列化的 EMR 集群和我们的本地机器之间确实不同，并且差异归因于 EMR 设置中存在的单个 class 而不是本地 - 这个 class 是 org.apache.spark.scheduler.HighlyCompressedMapStatus$CompressedSizes ，它不存在于任何公开可用的 Spark 代码中，因此我们将其归因于 Amazon spark 分支。这实际上意味着我们无法在本地读取几乎任何由 EMR 集群写入的 class，因为不可能在本地使用 Spark 的分支，并且 class 在创建时注册在 ID 13 Kryo 实例（以后可能会明显改变），导致几乎所有 classes 反序列化失败。
这里丑陋的修复是使用 Kryo 实例的 ClassResolver。如果注册表中不存在 CompressedSizes class，我们将所有 ID x >= 13 的 classes 注册为 x + 1。这确实很丑陋，但作为本地修复，它可以工作。显然，EMR/Kryo/Spark的新版本也可能会中断，所以要格外小心（我们只在本地使用它进行调试，这仍然很多）。
代码： 以前，我们会像这样创建 Kryo 实例：
val kryoSerializer = new KryoSerializer(sc.getConf)
val kryo = kryoSerializer.newKryo()
现在，我们使用这个：
val kryo = adjustRegistrationsForEmrSpark(kryoSerializer.newKryo())
哪里
private def adjustRegistrationsForEmrSpark(kryo: Kryo): Kryo = {
val existingRegistrations = getRegistrations(kryo)
val emrSpecificClassExists = existingRegistrations.exists(_.getType.getName.contains("CompressedSizes"))
if (emrSpecificClassExists) {
println(s"detected emr-specific class when creating kryo, not making any adjustments")
kryo
} else {
println(s"emr-specific class missing from registrations, adjusting existing classes by an offset of 1 to compensate")
val classResolver = kryo.getClassResolver
existingRegistrations.filter(_.getId >= 13).foreach { registration =>
val toRegister = new Registration(registration.getType, registration.getSerializer, registration.getId + 1)
classResolver.register(toRegister)
}
kryo
}
}
private def getRegistrations(kryo: Kryo): List[Registration] = {
var classIndex = 0
var reg: Registration = null
var result: List[Registration] = List()
do {
reg = kryo.getClassResolver.getRegistration(classIndex)
if (reg != null) result ++= List(reg)
classIndex = classIndex + 1
} while (reg != null)
result
}