如何使用 dask 分割大型 .csv 文件？

Question

我正在尝试使用它dask来将一个巨大的制表符分隔文件拆分为包含 100,000 个核心的 AWS Batch 阵列上的较小块。

在 AWS Batch 中，每个核心都有一个唯一的环境变量，AWS_BATCH_JOB_ARRAY_INDEX范围从 0 到 99,999（复制到idx下面代码片段中的变量中）。因此，我尝试使用以下代码：

import os
import dask.dataframe as dd

idx = int(os.environ["AWS_BATCH_JOB_ARRAY_INDEX"])

df = dd.read_csv(f"s3://main-bucket/workdir/huge_file.tsv", sep='\t')
df = df.repartition(npartitions=100_000)
df = df.partitions[idx]

df = df.persist() # this call isn't needed before calling to df.to_csv (see comment by Sultan)
df = df.compute() # this call isn't needed before calling to df.to_csv (see comment by Sultan)
df.to_csv(f"/tmp/split_{idx}.tsv", sep="\t", index=False)
print(idx, df.shape, df.head(5))

我需要先打电话presist和/或compute在打电话之前吗df.to_csv？

Answer 1

当我必须将一个大文件拆分为多个较小的文件时，我只需运行以下代码即可。

读取并重新分区

import dask.dataframe as dd

df = dd.read_csv("file.csv")
df = df.repartition(npartitions=100)

保存到 csv

o = df.to_csv("out_csv/part_*.csv", index=False)

保存到镶木地板

o = df.to_parquet("out_parquet/")

write_metadata_file=False如果您想避免元数据，可以使用此处。

几点注意事项：

• 我认为你真的不需要持久化和计算，因为你可以直接保存到磁盘。当遇到内存错误等问题时，保存到磁盘比计算更安全。
• 我发现在写入时使用 parquet 格式至少比 csv 快 3 倍。