如何将RDD.mapPartitions()返回的Pandas DataFrame转为Spark DataFrame？

你的问题核心在于mapPartitions的返回值不符合Spark创建DataFrame的要求，让我一步步帮你梳理和解决：

问题根源 #

你当前的func返回的是[pdf]——也就是每个分区返回一个包含Pandas DataFrame的列表。但Spark的spark.createDataFrame()期望的是一个包含行级对象（比如Spark的Row、元组）的可迭代序列，而不是把整个DataFrame当作单个元素来处理。当Spark尝试解析这个包含DataFrame的RDD时，Pandas会触发ValueError: The truth value of a DataFrame is ambiguous，因为它无法判断整个DataFrame作为一个"值"的真假性，这是Spark尝试将DataFrame当作单行数据处理时导致的。

解决方案：修改分区处理函数，返回行级迭代器 #

你需要把处理后的Pandas DataFrame拆成单个的行对象返回，让Spark能正确识别每一行数据。这里有两种常用的方式：

方式1：返回元组迭代器（最简洁）

利用Pandas的itertuples(index=False)方法，直接把DataFrame转换成元组的迭代器，每个元组对应一行数据，和输入的Row结构完全匹配：

import pandas as pd
from pyspark.sql import Row

def func(data):
    pdf = pd.DataFrame(list(data), columns=('A', 'B', 'C'))
    pdf += 10  # 你的业务逻辑
    # 返回元组迭代器，index=False去掉默认的行索引
    return pdf.itertuples(index=False)

# 原有代码保持不变，直到创建结果DataFrame
pdf = pd.DataFrame([(1.87, 0.6, 7.1), (-0.3, 0.1, 8.2), (2.8, 0.3, 6.1), (-0.2, 0.5, 5.9)], columns=('A', 'B', 'C'))
sdf = spark.createDataFrame(pdf)
sdf.show()

rddIn = sdf.rdd
result = rddIn.mapPartitions(func)
# 建议指定原schema，避免Spark自动推断时出错
resDf = spark.createDataFrame(result, schema=sdf.schema)
resDf.show()

方式2：返回Spark Row对象（类型更明确）

如果需要更明确的行类型，可以把Pandas的行转换成Spark的Row对象：

def func(data):
    pdf = pd.DataFrame(list(data), columns=('A', 'B', 'C'))
    pdf += 10
    # 遍历Pandas行，转换成Spark Row
    return [Row(**row.to_dict()) for _, row in pdf.iterrows()]

验证结果 #

修改后运行代码，resDf.show()会输出和输入结构完全一致的结果，只是每个数值都加了10：

+-----+----+----+
|    A|   B|   C|
+-----+----+----+
|11.87|10.6|17.1|
|  9.7|10.1|18.2|
| 12.8|10.3|16.1|
|  9.8|10.5|15.9|
+-----+----+----+

额外提示 #

对于复杂的业务逻辑，始终确保mapPartitions的处理函数返回的是单个行对象的可迭代序列，而不是整个DataFrame。另外，指定原DataFrame的schema可以避免Spark在自动推断schema时出现数据类型不匹配的问题，让代码更稳定。

内容的提问来源于stack exchange，提问作者snark