如果我调用 map 或 mapPartition 并且我的函数从 PySpark 接收行，那么创建本地 PySpark 或 Pandas DataFrame 的自然方法是什么？结合行并保留模式的东西？
目前我正在做类似的事情：
def combine(partition):
rows = [x for x in partition]
dfpart = pd.DataFrame(rows,columns=rows[0].keys())
pandafunc(dfpart)
mydf.mapPartition(combine)
【问题讨论】：
标签： python pandas apache-spark pyspark pyspark-sql
Spark >= 2.3.0
从 Spark 2.3.0 开始，可以按分区或组使用 Pandas Series 或 DataFrame。例如：
火花
创建本地 PySpark 的自然方式是什么
没有这样的事情。 Spark 分布式数据结构不能嵌套，或者您更喜欢不能嵌套动作或转换的另一个视角。
或 Pandas 数据帧
这相对容易，但你至少要记住几件事：
collections.OrderedDict 不同）。因此，传递列可能无法按预期工作。import pandas as pd
rdd = sc.parallelize([
{"x": 1, "y": -1},
{"x": -3, "y": 0},
{"x": -0, "y": 4}
])
def combine(iter):
rows = list(iter)
return [pd.DataFrame(rows)] if rows else []
rdd.mapPartitions(combine).first()
## x y
## 0 1 -1
【讨论】：
createDataFrame 并在那里传递架构（它不同于 Pandas dtypes）。
toPandas 简单地收集并创建与 Spark 数据帧具有相同 columns 名称的本地数据结构。不多也不少。 Row（如pyspark.sql.Row）没有架构——它只是一个tuple，添加了一些方法和__fields__存储名称的属性。
你可以使用toPandas()，
pandasdf = mydf.toPandas()
【讨论】：
map that passes dataframe。火花数据帧的输出预期是什么？您想为每个分区创建数据框吗？
实际上可以在执行器中将 Spark 行转换为 Pandas，并最终使用 mapPartitions 从这些输出中创建 Spark DataFrame。 See my gist in Github
# Convert function to use in mapPartitions
def rdd_to_pandas(rdd_):
# convert rows to dict
rows = (row_.asDict() for row_ in rdd_)
# create pandas dataframe
pdf = pd.DataFrame(rows)
# Rows/Pandas DF can be empty depending on patiition logic.
# Make sure to check it here, otherwise it will throw untrackable error
if len(pdf) > 0:
#
# Do something with pandas DataFrame
#
pass
return pdf.to_dict(orient='records')
# Create Spark DataFrame from resulting RDD
rdf = spark.createDataFrame(df.rdd.mapPartitions(rdd_to_pandas))
【讨论】：
为了创建 Spark SQL 数据框，您需要一个配置单元上下文：
hc = HiveContext(sparkContext)
使用 HiveContext，您可以通过 inferSchema 函数创建 SQL 数据框：
sparkSQLdataframe = hc.inferSchema(rows)
【讨论】：