pandas - 使用 Apache Arrow 将 PySpark DataFrame 转换为 Pandas

Question

我想将 PySpark DataFrame (pyspark.sql.DataFrame) 转换为 Pandas 数据帧。有一个内置方法 toPandas() 效率非常低(请阅读 Wes McKinney 在 Fenruary 2017 here 中关于此问题的文章以及他在 this jupyter notebook 中的计算)。

与此同时，我们已经做出了一些努力来加快此类转换的速度。一个例子是 Josh 的函数 here 。但是，这对我没有帮助，因为我希望将 +1M 行从 pysaprk.DataFrame 传输到 Pandas，这个解决方案对我不起作用。

幸运的是，如this post所示2017年7月26日，感谢作者Wes , Li和 Holden得益于 Spark 2.3 中 Apache Arrow 的实现，toPandas() 的功能得到了显着改进。话虽如此，我无法访问 Spark 2.3(我正在使用 Spark 2.1)。

所以，我的问题是如何使用 Apache Arrow 功能将 pyspark 数据帧快速转换为 Pandas 以供 Spark代码 > 2.1 之前的版本。我认为很多人都坚持使用旧版本的 Spark 并且可以从中受益。

更新 1:建议我先将 pyspark 打印到 CSV 文件，然后读取 CSV 文件来自 Pandas 强大的 read_csv 方法。我真的希望我能找到一种方法来避免这样做!

更新 2:toPandas() 方法缓慢的原因和可能的内存不足问题在 this discussion 中详细讨论。

dfSpark = spark.sql(sqlQuery)
df = dfSpark.toPandas() # Very slow / out of memory error

Answer 1

您尝试过使用中间文件吗？

您可以将文件从 Spark 保存到 Parquet，然后在 pandas 中读取。

#spark dataframe 
df.write.parquet("path/file.parquet")

查看更多: https://spark.apache.org/docs/2.1.0/sql-programming-guide.html#parquet-files

查看 pyarrow 读取 Parquet 文件:

https://arrow.apache.org/docs/python/parquet.html

import pyarrow.parquet as pq
table = pq.read_table('example.parquet') 

#or if you want to only read some of the colums 
table = pq.read_table('example.parquet', columns=['one', 'three'])

df = table.to_pandas()  #pandas df

此外，如果内存不足，请对 df 进行采样，或在写入之前对其进行过滤。