免责声明:刚开始玩 Spark。
我无法理解著名的“任务不可序列化”异常,但我的问题与我在 SO 上看到的问题有点不同(或者我认为如此)。
我有一个很小的自定义 RDD (TestRDD
)。它有一个字段,用于存储其类未实现可序列化 (NonSerializable
) 的对象。我已将“spark.serializer”配置选项设置为使用 Kryo。但是,当我在我的 RDD 上尝试 count()
时,我得到以下信息:
Caused by: java.io.NotSerializableException: com.complexible.spark.NonSerializable
Serialization stack:
- object not serializable (class: com.test.spark.NonSerializable, value: com.test.spark.NonSerializable@2901e052)
- field (class: com.test.spark.TestRDD, name: mNS, type: class com.test.spark.NonSerializable)
- object (class com.test.spark.TestRDD, TestRDD[1] at RDD at TestRDD.java:28)
- field (class: scala.Tuple2, name: _1, type: class java.lang.Object)
- object (class scala.Tuple2, (TestRDD[1] at RDD at TestRDD.java:28,<function2>))
at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
at org.apache.spark.scheduler.DAGScheduler.submitMissingTasks(DAGScheduler.scala:1009)
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$submitStage(DAGScheduler.scala:933)
当我查看 DAGScheduler.submitMissingTasks
时,我看到它在我的 RDD 上使用了它的 closure 序列化器,它是 Java 序列化器,而不是我想要的 Kryo 序列化器预计。我读过 Kryo 在序列化闭包方面存在问题,而 Spark 始终使用 Java 序列化程序来进行闭包,但我完全不明白闭包是如何在这里发挥作用的。我在这里所做的就是:
SparkConf conf = new SparkConf()
.setAppName("ScanTest")
.setMaster("local")
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
JavaSparkContext sc = new JavaSparkContext(conf);
TestRDD rdd = new TestRDD(sc.sc());
System.err.println(rdd.count());
也就是说,没有映射器或任何需要序列化闭包的东西。 OTOH 这行得通:
sc.parallelize(Arrays.asList(new NonSerializable(), new NonSerializable())).count()
Kryo 序列化器按预期使用,不涉及闭包序列化器。如果我没有将序列化程序属性设置为 Kryo,我也会在此处遇到异常。
我感谢任何解释闭包来源以及如何确保我可以使用 Kryo 序列化自定义 RDD 的指示。
更新:这是带有不可序列化字段mNS
的TestRDD
:
class TestRDD extends RDD<String> {
private static final ClassTag<String> STRING_TAG = ClassManifestFactory$.MODULE$.fromClass(String.class);
NonSerializable mNS = new NonSerializable();
public TestRDD(final SparkContext _sc) {
super(_sc,
JavaConversions.asScalaBuffer(Collections.<Dependency<?>>emptyList()),
STRING_TAG);
}
@Override
public Iterator<String> compute(final Partition thePartition, final TaskContext theTaskContext) {
return JavaConverters.asScalaIteratorConverter(Arrays.asList("test_" + thePartition.index(),
"test_" + thePartition.index(),
"test_" + thePartition.index()).iterator()).asScala();
}
@Override
public Partition[] getPartitions() {
return new Partition[] {new TestPartition(0), new TestPartition(1), new TestPartition(2)};
}
static class TestPartition implements Partition {
final int mIndex;
public TestPartition(final int theIndex) {
mIndex = theIndex;
}
public int index() {
return mIndex;
}
}
}
最佳答案
When I look inside
DAGScheduler.submitMissingTasks
I see that it uses its closure serializer on my RDD, which is the Java serializer, not the Kryo serializer which I'd expect.
SparkEnv
支持两种序列化器,一种名为serializer
,用于数据序列化、检查点、工作人员之间的消息传递等,可在 spark 下使用。 serializer
配置标志。另一个称为 spark.closure.serializer
下的 closureSerializer
,用于检查您的对象实际上是可序列化的并且可配置为 Spark <= 1.6.2(但没有除了 JavaSerializer
实际工作之外)并从 2.0.0 及更高版本硬编码到 JavaSerializer
。
Kryo 闭包序列化程序有一个错误导致它无法使用,您可以在 SPARK-7708 下查看该错误(这可能已通过 Kryo 3.0.0 修复,但 Spark 目前已通过特定版本的 Chill 修复,该版本已在 Kryo 2.2.1 上修复)。此外,对于 Spark 2.0.x,JavaSerializer 现在是固定的而不是可配置的(您可以看到它 in this pull request )。这意味着实际上我们只能使用 JavaSerializer
来进行闭包序列化。
我们使用一个序列化器来提交任务,而另一个序列化器在工作人员之间序列化数据,这很奇怪吗?当然可以,但这就是我们所拥有的。
总而言之,如果您正在设置 spark.serializer
配置,或使用 SparkContext.registerKryoClasses
,您将在 Spark 中使用 Kryo 进行大部分序列化.话虽如此,为了检查给定类是否可序列化并将任务序列化给工作人员,Spark 将使用 JavaSerializer
。
关于java - 了解 Spark 的闭包及其序列化,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/40259196/