好吧,细节在于魔鬼。要了解发生这种情况的原因,我们将不得不仔细研究PySpark序列化器。首先让我们
SparkContext使用默认设置进行创建:
from pyspark import SparkContextsc = SparkContext("local", "foo")
并检查什么是默认的序列化器:
sc.serializer## AutoBatchedSerializer(PickleSerializer())sc.serializer.bestSize## 65536
它告诉我们三件事:
- 这是
AutoBatchedSerializer
序列化器 - 它
PickleSerializer
用来执行实际的工作 bestSize
批处理的序列化的是65536字节
快速浏览一下源代码,您会发现此序列化调整了运行时时间序列化的记录数,并尝试使批大小保持小于10*
bestSize。重要的一点是,并非单个分区中的所有记录都同时被序列化。
我们可以通过以下实验进行检查:
from operator import addbd = sc.broadcast({})rdd = sc.parallelize(range(10), 1).map(lambda _: bd.value)rdd.map(id).distinct().count()## 1rdd.cache().count()## 10rdd.map(id).distinct().count()## 2
如您所见,即使在序列化-反序列化之后的这个简单示例中,我们也得到了两个不同的对象。您可以直接使用观察到类似的行为
pickle:
v = {}vs = [v, v, v, v]v1, *_, v4 = pickle.loads(pickle.dumps(vs))v1 is v4## True(v1_, v2_), (v3_, v4_) = ( pickle.loads(pickle.dumps(vs[:2])), pickle.loads(pickle.dumps(vs[2:])))v1_ is v4_## Falsev3_ is v4_## True
取消酸洗后,在同一批引用中序列化的值是同一对象。来自不同批次的值指向不同的对象。
在实践中,Spark具有多个序列化和不同的序列化策略。例如,您可以使用无限大小的批次:
from pyspark.serializers import BatchedSerializer, PickleSerializerrdd_ = (sc.parallelize(range(10), 1).map(lambda _: bd.value) ._reserialize(BatchedSerializer(PickleSerializer())))rdd_.cache().count()rdd_.map(id).distinct().count()## 1
您可以通过将
serializer和/或
batchSize参数传递给
SparkContext构造函数来更改序列化程序:
sc = SparkContext( "local", "bar", serializer=PickleSerializer(), # Default serializer # Unlimited batch size -> BatchedSerializer instead of AutoBatchedSerializer batchSize=-1 )sc.serializer## BatchedSerializer(PickleSerializer(), -1)
选择不同的序列化程序和批处理策略会导致不同的权衡(速度,序列化任意对象的能力,内存要求等)。
您还应该记住,Spark中的广播变量不会在执行程序线程之间共享,因此同一工作线程可以同时存在多个反序列化副本。
此外,如果执行需要改组的转换,您将看到与此类似的行为。