时间戳

首页 > 解决方案 > 如何使用 dask 拆分大型 .csv 文件？

python - 如何使用 dask 拆分大型 .csv 文件？

问题描述

我正在尝试使用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吗？

标签： pythoncsvdaskaws-batch

解决方案

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

读取和重新分区

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 倍。

推荐阅读