使用Dask分块高效读取大型CSV文件

Question

现在，我正在使用Dask读取大型CSV文件，并对其进行一些后处理（例如执行一些数学运算，然后通过某些ML模型进行预测并将结果写入数据库）。为避免加载所有数据到内存中，我想按当前大小读取数据块：读取第一个数据块，进行预测，写入数据，读取第二个数据块等。

我尝试了使用skiprows和nrows的解决方案：

import dask.dataframe as dd
read_path = "medium.csv"

# Read by chunk
skiprows = 100000
nrows = 50000
res_df = dd.read_csv(read_path, skiprows=skiprows)
res_df = res_df.head(nrows)

print(res_df.shape)
print(res_df.head())

但是我遇到了错误：

ValueError：样本不足以包含至少一行数据。请在调用read_csv/read_table时增加sample中的字节数。

而且，据我所知，它将为所有数据计算二进制掩码（[False，False，...，True，...]）以查找要加载的行。我们如何更有效地实现它？也许使用dask的分布式或延迟函数？

- Mikhail_Sam

Answer 1

Dask DataFrame会为你分割数据，你不需要使用nrows/skip_rows

df = dd.read_csv(filename)

如果您想选择特定的分区，则可以使用分区访问器。

part = df.partitions[i]

然而，您可能还希望并行应用您的功能。

df.map_partitions(process).to_csv("data.*.csv")

- MRocklin