我正在使用 pandas 和 spark 数据帧。数据帧总是非常大(> 20 GB),标准的 spark 函数不足以满足这些大小。目前我正在将我的 pandas 数据框转换为这样的 spark 数据框:
dataframe = spark.createDataFrame(pandas_dataframe)
我进行这种转换是因为使用 spark 将数据帧写入 hdfs 非常容易:
dataframe.write.parquet(output_uri, mode="overwrite", compression="snappy")
但是对于大于 2 GB 的数据帧,转换失败。 如果我将 spark 数据框转换为 pandas,我可以使用 pyarrow:
// temporary write spark dataframe to hdfs
dataframe.write.parquet(path, mode="overwrite", compression="snappy")
// open hdfs connection using pyarrow (pa)
hdfs = pa.hdfs.connect("default", 0)
// read parquet (pyarrow.parquet (pq))
parquet = pq.ParquetDataset(path_hdfs, filesystem=hdfs)
table = parquet.read(nthreads=4)
// transform table to pandas
pandas = table.to_pandas(nthreads=4)
// delete temp files
hdfs.delete(path, recursive=True)
这是从 spark 到 pandas 的快速转换,它也适用于大于 2 GB 的数据帧。我还找不到相反的方法。意思是有一个 pandas 数据框,我在 pyarrow 的帮助下将其转换为 spark。问题是我真的找不到如何将 pandas 数据帧写入 hdfs。
我的 Pandas 版本:0.19.0
最佳答案
Meaning having a pandas dataframe which I transform to spark with the help of pyarrow.
pyarrow.Table.fromPandas
是您正在寻找的功能:
Table.from_pandas(type cls, df, bool timestamps_to_ms=False, Schema schema=None, bool preserve_index=True) Convert pandas.DataFrame to an Arrow Table
import pyarrow as pa
pdf = ... # type: pandas.core.frame.DataFrame
adf = pa.Table.from_pandas(pdf) # type: pyarrow.lib.Table
结果可以直接写入Parquet/HDFS,无需通过Spark传递数据:
import pyarrow.parquet as pq
fs = pa.hdfs.connect()
with fs.open(path, "wb") as fw
pq.write_table(adf, fw)
另见
- @WesMcKinney answer至 read a parquet files from HDFS using PyArrow .
- Reading and Writing the Apache Parquet Format在
pyarrow
documentation . - Native Hadoop file system (HDFS) connectivity in Python
Spark 笔记:
此外,自 Spark 2.3(当前主版本)以来,createDataFrame
(SPARK-20791 - Use Apache Arrow to Improve Spark createDataFrame from Pandas.DataFrame)直接支持 Arrow。它uses SparkContext.defaultParallelism
to compute number of chunks这样您就可以轻松控制各个批处理的大小。
最后,defaultParallelism
可用于控制使用标准 _convert_from_pandas
生成的分区数量,有效地将切片的大小减小到更易于管理的程度。
不幸的是,这些不太可能解决您的 current memory problems .两者都依赖于parallelize
,因此将所有数据存储在驱动程序节点的内存中。切换到 Arrow 或调整配置只能加快进程或解决 block 大小限制。
实际上,只要您使用本地 Pandas DataFrame
作为输入,我看不出有任何理由在这里切换到 Spark。这种情况下最严重的瓶颈是驱动程序的网络 I/O,而分发数据无法解决这个问题。
关于python - 如何将巨大的 Pandas 数据框保存到 hdfs?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/49194737/