Spark DataFrame mapPartitions

Posted 2023-04-13

【中文标题】Spark DataFrame mapPartitions

【英文标题】：

【发布时间】：2016-08-03 16:34:03

【问题描述】：

我需要对 Spark DataFrame 进行分布式计算，在 DataFrame 块上调用一些任意（不是 SQL）逻辑。 我做到了：

def some_func(df_chunk):
    pan_df = df_chunk.toPandas()
    #whatever logic here

df = sqlContext.read.parquet(...)
result = df.mapPartitions(some_func)

不幸的是，它导致：

AttributeError: 'itertools.chain' 对象没有属性 'toPandas'

我希望在每个地图调用中都有 spark DataFrame 对象，而不是我得到了“itertools.chain”。为什么？又该如何克服呢？

【问题讨论】：

PySpark 正在使用 itertools.chain 将数据传递给 mapPartition，因此您将此对象传递给它无法识别的函数。

【参考方案1】：

试试这个：

>>> columns = df.columns
>>> df.rdd.mapPartitions(lambda iter: [pd.DataFrame(list(iter), columns=columns)])

【讨论】：

“结果”在哪里？你如何在 OP 中调用 some_func？