python - 为什么 1 行 DataFrame 上的 collect() 使用 2000 个执行器？

Question

这是我能想到的最简单的 DataFrame。我正在使用 PySpark 1.6.1。

# one row of data
rows = [ (1,   2) ]
cols = [ "a", "b" ]
df   = sqlContext.createDataFrame(rows, cols)

所以数据框完全适合内存，没有对任何文件的引用，对我来说看起来很微不足道。

然而，当我收集数据时，它使用了 2000 个执行程序:

df.collect()

在收集过程中，使用了 2000 个执行器:

[Stage 2:===================================================>(1985 + 15) / 2000]

然后是预期的输出:

[Row(a=1, b=2)]

为什么会这样？ DataFrame 不应该完全在驱动程序的内存中吗？

Answer 1

所以我仔细研究了代码，试图弄清楚发生了什么。看起来 sqlContext.createDataFrame 确实没有尝试根据数据设置合理的参数值。

为什么有 2000 个任务？

Spark 使用 2000 个任务，因为我的数据框有 2000 个分区。 (尽管分区多于行似乎是无稽之谈。)

这可以通过以下方式看到:

>>> df.rdd.getNumPartitions()
2000

为什么 DataFrame 有 2000 个分区？

发生这种情况是因为 sqlContext.createDataFrame 最终使用默认分区数(在我的例子中为 2000)，而不管数据是如何组织的或者它有多少行。

代码轨迹如下。

在 sql/context.py 中，sqlContext.createDataFrame 函数调用(在本例中):

rdd, schema = self._createFromLocal(data, schema)

依次调用:

return self._sc.parallelize(data), schema

sqlContext.parallelize函数定义在context.py中:

numSlices = int(numSlices) if numSlices is not None else self.defaultParallelism

没有对行数进行检查，也无法从 sqlContext.createDataFrame 中指定切片数。

如何更改 DataFrame 的分区数？

使用 DataFrame.coalesce。

>>> smdf = df.coalesce(1)
>>> smdf.rdd.getNumPartitions()
1
>>> smdf.explain()
== Physical Plan ==
Coalesce 1
+- Scan ExistingRDD[a#0L,b#1L]
>>> smdf.collect()
[Row(a=1, b=2)]