将pandas的df.to_parquet写入多个较小文件

Question

是否可以使用Pandas的DataFrame.to_parquet功能将写入拆分为多个大约所需大小的文件？

我有一个非常大的DataFrame（100M x 100），并且正在使用df.to_parquet（'data.snappy'，engine ='pyarrow'，compression ='snappy'）写入文件，但这会导致一个大约4GB的文件。 我想将其拆分成许多约100MB的文件。

- Austin

---

请执行与“Pandas dataframe to_csv-拆分为多个输出文件”相同的操作，只不过使用.to_parquet()进行操作。 - Trenton McKinney

最终使用了ddf = dask.dataframe.from_pandas(df, chunksize=5000000); ddf.to_parquet('/path/to/save/')，每个块保存为一个文件。 - Austin

你应该将其写成答案。这可能对他人有益。 - Trenton McKinney

Answer 1

我最终使用了Dask：

import dask.dataframe as da

ddf = da.from_pandas(df, chunksize=5000000)
save_dir = '/path/to/save/'
ddf.to_parquet(save_dir)

这将保存到save_dir内的多个Parquet文件中，其中每个子数据帧的行数为chunksize。根据您的数据类型和列数，您可以调整此选项以获取所需大小的文件。

- Austin

Answer 2

另一种选择是在pyarrow.parquet.write_to_dataset()中使用partition_cols选项:

import pyarrow.parquet as pq
import numpy as np

# df is your dataframe
n_partition = 100
df["partition_idx"] = np.random.choice(range(n_partition), size=df.shape[0])
table = pq.Table.from_pandas(df, preserve_index=False)
pq.write_to_dataset(table, root_path="{path to dir}/", partition_cols=["partition_idx"])

- Random Certainty

Answer 3

使用仅限于pandas api（不使用dask或pyarrow）将数据框切片并将每个块保存到文件夹中。

如果需要，可以向parquet引擎传递额外的参数。

def df_to_parquet(df, target_dir, chunk_size=1000000, **parquet_wargs):
"""Writes pandas DataFrame to parquet format with pyarrow.

Args:
    df: DataFrame
    target_dir: local directory where parquet files are written to
    chunk_size: number of rows stored in one chunk of parquet file. Defaults to 1000000.
"""    
for i in range(0, len(df), chunk_size):
    slc = df.iloc[i : i + chunk_size]
    chunk = int(i/chunk_size)
    fname = os.path.join(target_dir, f"part_{chunk:04d}.parquet")
    slc.to_parquet(fname, engine="pyarrow", **parquet_wargs)

- Maciej Skorski

Answer 4

保持每个Parquet文件大小小，大约在128MB左右。为此，请执行以下操作：

import dask.dataframe as dd

# Get number of partitions required for nominal 128MB partition size
# "+ 1" for non full partition
size128MB = int(df.memory_usage().sum()/1e6/128) + 1
# Read
ddf = dd.from_pandas(df, npartitions=size128MB)
save_dir = '/path/to/save/'
ddf.to_parquet(save_dir)

- winderland

Answer 5

cunk = 200000
    i = 0
    n = 0
    while i<= len(all_df):
        j = i + cunk
        print((i, j))
        tmpdf = all_df[i:j]
        tmpdf.to_parquet(path=f"./append_data/part.{n}.parquet",engine='pyarrow', compression='snappy')
        i = j
        n = n + 1

- L bear

---

请添加一个简要的解释，说明它如何/为什么解决了问题。 - rachwa