现在,我正在使用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的分布式或延迟函数?