Spark using python: How to resolve Stage x 包含一个非常大的任务 (xxx KB)。建议的最大任务大小为 100 KB答案

【问题标题】：Spark using python: How to resolve Stage x contains a task of very large size (xxx KB). The maximum recommended task size is 100 KBSpark using python: How to resolve Stage x 包含一个非常大的任务 (xxx KB)。建议的最大任务大小为 100 KB
【发布时间】：2015-05-06 20:52:23
【问题描述】：

我刚刚创建了 range(1,100000) 的 python 列表。

使用 SparkContext 完成以下步骤：

a = sc.parallelize([i for i in range(1, 100000)])
b = sc.parallelize([i for i in range(1, 100000)])

c = a.zip(b)

>>> [(1, 1), (2, 2), -----]

sum  = sc.accumulator(0)

c.foreach(lambda (x, y): life.add((y-x)))

警告如下：

ARN TaskSetManager：第 3 阶段包含一个非常大的任务 (4644 KB)。建议的最大任务大小为 100 KB。

如何解决此警告？有没有办法处理大小？还有，会不会影响大数据的时间复杂度？

【问题讨论】：

首先life到底是什么？你是说sum 累加器吗？不管怎样，这里不应该是个问题。详情请见this。

标签： apache-spark spark-streaming

【解决方案1】：

一般的想法是，PySpark 创建的 java 进程数与 executor 数一样多，然后将数据传送到每个进程。如果进程太少，java堆空间就会出现内存瓶颈。

在您的情况下，具体错误是您使用sc.parallelize([...]) 创建的RDD 未指定分区数（参数numSlices，请参阅docs）。而RDD默认的分区数太小了（有可能是由单个分区构成的）。

要解决这个问题，只需指定想要的分区数：

a = sc.parallelize([...], numSlices=1000)   # and likewise for b

随着您指定的切片数量越来越多，您将看到警告消息中所述的大小减小。增加切片的数量，直到您不再收到警告消息。例如，得到

Stage 0 contains a task of very large size (696 KB). The maximum recommended task size is 100 KB

表示需要指定更多的切片。

在处理内存问题时可能有用的另一个提示（但这与警告消息无关）：默认情况下，每个执行程序可用的内存为 1 GB 左右。您可以通过命令行指定更大的数量，例如使用--executor-memory 64G。

【讨论】：

如何确定要使用的最佳切片数？从您的示例来看，似乎需要 7 个将每个保持在 100 KB 以下。虽然我如何确定任务的大小是多少？
@user2361174 正如您所写，您将通过警告消息 (blabla contains a task of very large size..) 知道大小是否太高 - 在尝试之前我不知道有任何其他方法可以获取此信息。
如果你 import sys, math 则 n = math.ceil(sys.getsizeof(your_list) / 102400) 将是保持所有切片小于 100KB 的最小切片数

【解决方案2】：

Spark 在发送任务期间会原生发送每个变量的副本。对于大尺寸的此类变量，您可能需要使用Broadcast Variables

如果您仍然面临大小问题，那么也许这些数据本身应该是一个 RDD

【讨论】：

嗨@Hitesh Dharamdasani，有什么方法可以更改SPARK 配置文件以最大化任务大小。？
对不起，我迟到了。 spark.apache.org/docs/1.2.0/tuning.html 在“数据序列化”部分有一些建议。但一般不推荐。我在 kyro 序列化程序上取得了一些成功，但没有什么可吹嘘的。广播变量更好
如果变量不可序列化怎么办（因此你必须使用像mapPartitions这样的运算符）？我不认为它可以包装在广播变量中。
只是为了进一步澄清这个答案。在提供的示例中超过 100 KB 的变量是 [i for i in range(1, 100000)]

【解决方案3】：

扩展@leo9r 评论：考虑不要使用python range，而是使用sc.range https://spark.apache.org/docs/1.6.0/api/python/pyspark.html#pyspark.SparkContext.range。

因此，您可以避免将大量列表从驱动程序转移到执行程序。

当然，此类 RDD 通常仅用于测试目的，因此您不希望它们被广播。

【讨论】：

使用 sc.range 而不是 range 在玩具示例中有效，但错过了更普遍的问题（python 和 java 之间如何通信数据）