map_partitions 的返回值是多少？

Question

dask API 说，map_partition 可用于“在每个 DataFrame 分区上应用 Python 函数”。根据这个描述并根据“map”的通常行为，我希望 map_partitions 的返回值是（类似于）一个长度等于分区数的列表。列表的每个元素都应该是函数调用的返回值之一。

但是，关于下面的代码，我不确定，返回值取决于什么：

#generate example dataframe
pdf = pd.DataFrame(np.random.randint(0,100,size=(100, 4)), columns=list('ABCD'))
ddf = dd.from_pandas(pdf, npartitions=3)

#define helper function for map. VAL is the return value
VAL = pd.Series({'A': 1})
#VAL = pd.DataFrame({'A': [1]}) #other return values used in this example
#VAL = None
#VAL = 1
def helper(x):
    print('function called\n')
    return VAL

#check result
out = ddf.map_partitions(helper).compute()
print(len(out))

• VAL = pd.Series({'A': 1}) 导致 4 个函数调用（可能一个用于推断 dtype，3 个用于分区）和一个 len == 3 和 pd.Series 类型的输出。
• pd.DataFrame({'A': [1]}) 产生相同的数字，但结果类型是 pd.DataFrame。
• VAL = None 导致 TypeError ... 为什么？难道 map_partitions 的一种可能用途是做某事而不是返回某事吗？
• VAL = 1 仅导致 2 个函数调用。 map_partitions 的结果是整数 1。

因此，我想问一些问题：

1. map_partitions 的返回值是如何确定的？
2. 除了分区数量之外，还有什么影响函数调用的数量/每个分区调用一次函数需要满足什么条件？
3. 一个函数的返回值应该是什么，它只是“做”某事，即一个过程？
4. 应该如何设计一个返回任意对象的函数？

Answer 1

Dask DataFrame.map_partitions 函数根据映射函数的输出类型返回一个新的 Dask Dataframe 或 Series。有关详细说明，请参阅API documentation。

1. map_partitions的返回值是如何确定的？

   请参阅上面提到的 API 文档。

2. 除了分区数量之外，还有什么影响函数调用的数量/每个分区调用一次函数需要满足什么条件？

   您是正确的，我们立即调用它一次来猜测输出的 dtypes/columns。您可以通过直接指定 meta= 关键字来避免这种情况。除此之外，该函数每个分区调用一次。

3. 一个函数的返回值应该是什么，它只“做”某事，即一个过程？

   您总是可以返回一个空的数据框。您可能还需要考虑将数据框转换为一系列 dask.delayed 对象，这些对象通常更常用于临时计算。

4. 应该如何设计一个返回任意对象的函数？

   如果您的函数不返回系列/数据帧，那么我建议您使用 DataFrame.to_delayed 方法将您的数据帧转换为 dask.delayed 对象序列。