python - 正确使用大广播变量的技巧？

Question

我正在使用大小约为 100 MB 的广播变量，我将其近似为:

>>> data = list(range(int(10*1e6)))
>>> import cPickle as pickle
>>> len(pickle.dumps(data))
98888896

在具有 3 个 c3.2xlarge 执行程序和一个 m3.large 驱动程序的集群上运行，使用以下命令启动交互式 session :

IPYTHON=1 pyspark --executor-memory 10G --driver-memory 5G --conf spark.driver.maxResultSize=5g

在 RDD 中，如果我持久化对这个广播变量的引用，内存使用量就会激增。对于 100 MB 变量的 100 次引用，即使它被复制 100 次，我预计数据使用总量也不会超过 10 GB(更不用说 3 个节点上的 30 GB 了)。但是，当我运行以下测试时，我看到内存不足的错误:

data = list(range(int(10*1e6)))
metadata = sc.broadcast(data)
ids = sc.parallelize(zip(range(100), range(100)))
joined_rdd = ids.mapValues(lambda _: metadata.value)
joined_rdd.persist()
print('count: {}'.format(joined_rdd.count()))

堆栈跟踪:

TaskSetManager: Lost task 17.3 in stage 0.0 (TID 75, 10.22.10.13): 

org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/worker.py", line 111, in main
    process()
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/worker.py", line 106, in process
    serializer.dump_stream(func(split_index, iterator), outfile)
  File "/usr/lib/spark/python/pyspark/rdd.py", line 2355, in pipeline_func
    return func(split, prev_func(split, iterator))
  File "/usr/lib/spark/python/pyspark/rdd.py", line 2355, in pipeline_func
    return func(split, prev_func(split, iterator))
  File "/usr/lib/spark/python/pyspark/rdd.py", line 317, in func
    return f(iterator)
  File "/usr/lib/spark/python/pyspark/rdd.py", line 1006, in <lambda>
    return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()
  File "/usr/lib/spark/python/pyspark/rdd.py", line 1006, in <genexpr>
    return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 139, in load_stream
    yield self._read_with_length(stream)
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 164, in _read_with_length
    return self.loads(obj)
  File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 422, in loads
    return pickle.loads(obj)
MemoryError


  at org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:138)
  at org.apache.spark.api.python.PythonRDD$$anon$1.<init>(PythonRDD.scala:179)
  at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:97)
  at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)
  at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
  at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
  at org.apache.spark.scheduler.Task.run(Task.scala:88)
  at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
  at org.apache.spark.scheduler.Task.run(Task.scala:88)
  at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
  at java.lang.Thread.run(Thread.java:745)

16/05/25 23:57:15 ERROR TaskSetManager: Task 17 in stage 0.0 failed 4 times; aborting job
---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
<ipython-input-1-7a262fdfa561> in <module>()
      7 joined_rdd.persist()
      8 print('persist called')
----> 9 print('count: {}'.format(joined_rdd.count()))

/usr/lib/spark/python/pyspark/rdd.py in count(self)
   1004         3
   1005         """
-> 1006         return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()
   1007
   1008     def stats(self):

/usr/lib/spark/python/pyspark/rdd.py in sum(self)
    995         6.0
    996         """
--> 997         return self.mapPartitions(lambda x: [sum(x)]).fold(0, operator.add)
    998
    999     def count(self):

/usr/lib/spark/python/pyspark/rdd.py in fold(self, zeroValue, op)
    869         # zeroValue provided to each partition is unique from the one provided
    870         # to the final reduce call
--> 871         vals = self.mapPartitions(func).collect()
    872         return reduce(op, vals, zeroValue)
    873

/usr/lib/spark/python/pyspark/rdd.py in collect(self)
    771         """
    772         with SCCallSiteSync(self.context) as css:
--> 773             port = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())
    774         return list(_load_from_socket(port, self._jrdd_deserializer))
    775

/usr/lib/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py in __call__(self, *args)

  at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
  at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)
  at py4j.Gateway.invoke(Gateway.java:259)
  at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
  at py4j.commands.CallCommand.execute(CallCommand.java:79)
  at py4j.GatewayConnection.run(GatewayConnection.java:207)
  at java.lang.Thread.run(Thread.java:745)

我已经看到以前的线程关于 pickle 反序列化的内存使用是一个问题。但是，我希望广播变量仅被反序列化(并在执行程序上加载到内存中)一次，随后对 .value 的引用将引用该内存地址。然而，情况似乎并非如此。我错过了什么吗？

我看到的广播变量示例将它们作为字典，用于一次转换一组数据(即用机场名称替换机场首字母缩写词)。将它们保留在这里的动机是创建具有广播变量知识的对象以及如何与之交互、保留这些对象并使用它们执行多项计算(spark 负责将它们保存在内存中)。

使用大型(100 MB 以上)广播变量有哪些技巧？坚持广播变量是否被误导？这可能是 PySpark 特有的问题吗？

谢谢!感谢您的帮助。

请注意，我还在 databricks forums 上发布了这个问题

编辑 - 后续问题:

建议默认的 Spark 序列化器的批处理大小为 65337。在不同批处理中序列化的对象不会被识别为相同的，并且会分配不同的内存地址，通过内置的 id 函数检查.然而，即使使用理论上需要 256 批序列化的更大广播变量，我仍然只能看到 2 个不同的副本。我不应该看更多吗？我对批量序列化工作原理的理解有误吗？

>>> sc.serializer.bestSize
65536
>>> import cPickle as pickle
>>> broadcast_data = {k: v for (k, v) in enumerate(range(int(1e6)))}
>>> len(pickle.dumps(broadcast_data))
16777786
>>> len(pickle.dumps({k: v for (k, v) in enumerate(range(int(1e6)))})) / sc.serializer.bestSize
256
>>> bd = sc.broadcast(broadcast_data)
>>> rdd = sc.parallelize(range(100), 1).map(lambda _: bd.value)
>>> rdd.map(id).distinct().count()
1
>>> rdd.cache().count()
100
>>> rdd.map(id).distinct().count()
2

Answer 1

嗯，细节决定成败。要了解发生这种情况的原因，我们必须仔细研究 PySpark 序列化程序。首先让我们使用默认设置创建 SparkContext:

from pyspark import SparkContext

sc = SparkContext("local", "foo")

并检查什么是默认序列化程序:

sc.serializer
## AutoBatchedSerializer(PickleSerializer())

sc.serializer.bestSize
## 65536

它告诉我们三个不同的事情:

• 这是AutoBatchedSerializer序列化器
• 它正在使用 PickleSerializer 来执行实际工作
序列化批处理的
• bestSize为65536字节

快速浏览 at the source code将向您展示此序列化在运行时调整当时序列化的记录数，并尝试使批处理大小小于 10 * bestSize。重要的一点是，并非单个分区中的所有记录都同时序列化。

我们可以通过如下实验验证:

from operator import add

bd = sc.broadcast({})

rdd = sc.parallelize(range(10), 1).map(lambda _: bd.value)
rdd.map(id).distinct().count()
## 1

rdd.cache().count()
## 10

rdd.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_
## False

v3_ is v4_
## True

在同一个批处理引用中序列化的值，在 unpickling 之后，同一个对象。来自不同批处理的值指向不同的对象。

在实践中，Spark 有多个序列化和不同的序列化策略。例如，您可以使用无限大小的批处理:

from pyspark.serializers import BatchedSerializer, PickleSerializer

rdd_ = (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 中的广播变量不会在执行线程之间共享，因此同一个 worker 可以同时存在多个反序列化副本。

此外，如果您执行需要改组的转换，您会看到与此类似的行为。