如何在 numpy/pandas 中处理来自 spark 的大型镶木地板文件？答案

【问题标题】：How can I process a large parquet file from spark in numpy/pandas?如何在 numpy/pandas 中处理来自 spark 的大型镶木地板文件？
【发布时间】：2021-09-03 23:56:09
【问题描述】：

我为 pandas、numpy 和 spark 标签发布此内容，因为我不确定在这三个系统中解决此问题的最佳方法。

我有一个大型 parquet 文件，下游进程无法打开它，因为它超出了系统的内存（如果一次打开，内存中约 63gb）。我是这样写文件的：

FULL_MAIN.write.mode("overwrite").parquet(PATH+"/FULL_MAIN.parquet")

但是文件太大了，所以我尝试这样做以将文件分成更小的夹头：

    split_factor = [.1,.1,.1,.1,.1,.1,.1,.1,.1,.1]
    FULL_MAIN_RDD1,FULL_MAIN_RDD2,FULL_MAIN_RDD3,FULL_MAIN_RDD4,FULL_MAIN_RDD5, FULL_MAIN_RDD6,FULL_MAIN_RDD7,FULL_MAIN_RDD8,FULL_MAIN_RDD9,FULL_MAIN_RDD10  = FULL_MAIN.randomSplit(split_factor)
FULL_MAIN_RDD1.write.mode("overwrite").parquet(PATH+"/FULL_MAIN_RDD1.parquet")
FULL_MAIN_RDD2.write.mode("overwrite").parquet(PATH+"/FULL_MAIN_RDD2.parquet")
...

这种方法的问题是我需要其他数据帧来保持行对齐，并且进行这种随机拆分会使数据帧不对齐。

所以我的两个问题是：

当我的数据集中的每一行都没有任何行号或数字计数器时，是否可以按相对相等的数量拆分多个数据帧？
有没有办法在 pandas 或 numpy 中批量读取 parquet 文件？这基本上可以解决我在下游系统上的问题。我不知道如何分批打开镶木地板（我试图在 pandas 中打开它，然后拆分行并保存每个文件，但是当我加载数据框时它会导致我的系统崩溃）。我不确定是否可以不超出内存。

【问题讨论】：

你可以看看这个链接stackoverflow.com/questions/59098785/…

标签： pandas numpy apache-spark pyspark parquet

【解决方案1】：

Parquet 文件格式支持行组。安装pyarrow并在创建parquet文件时使用row_groups：

df.to_parquet("filename.parquet", row_group_size=10000, engine="pyarrow")

然后您可以逐组（甚至仅特定组）阅读：

import pyarrow.parquet as pq

pq_file = pq.ParquetFile("filename.parquet")
n_groups = pq_file.num_row_groups
for grp_idx in range(n_groups):
    df = pq_file.read_row_group(grp_idx, use_pandas_metadata=True).to_pandas()
    process(df)

如果您无法控制 parquet 文件的创建，您仍然可以只读取文件的一部分：

pq_file = pq.ParquetFile("filename.parquet")
batch_size = 10000 # records

batches = pq_file.iter_batches(batch_size, use_pandas_metadata=True) # batches will be a generator    
for batch in batches:
    df = batch.to_pandas()
    process(df)

【讨论】：

【解决方案2】：

我不确定你是否有 spark 。如果您想提供下游较小的文件块，您可以使用重新分区到所需数量的块并重写 parquet 文件。您可以根据需要更改重新分区号。

df = spark.read.parquet('filename.parquet')
df.repartition(200).mode('overwrite').save('targetPath')

【讨论】：