gpt4 book ai didi

java - 了解 Spark 的闭包及其序列化

转载 作者:太空狗 更新时间:2023-10-29 22:41:22 25 4
gpt4 key购买 nike

免责声明:刚开始玩 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 的指示。

更新:这是带有不可序列化字段mNSTestRDD:

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/

25 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com