Dask Distributed on local machine metadata column order mismatch with smaller partition sizes

Dask Distributed on local machine metadata column order mismatch with smaller partition sizes

当我尝试使用 dask 清理一些 jsonl 文件时，如果我使用大约 128MB 的分区大小，它会给我错误提示列顺序不匹配，但如果我使用时有点慢，则工作正常大小为 512MB

所以我的代码看起来像：

import dask.dataframe as dd
import pandas as pd
from dask.distributed import Client
client = Client(n_workers = 1, threads_per_worker = 4)
meta = dd.utils.make_meta([('x1', 'object'), ('date', 'int64'), ('x2', 'object'])
df = dd.read_json('*.jsonl', blocksize = 2**27, meta = meta)
keep = ['item1', 'item2', 'item3']
df['x1'] = df.x1.str.lower()
df = df[df['x1'].isin(keep)]
df.to_csv('dask_file.csv', single_file) = True

当我运行这段代码时，我最终得到一个错误，指出列的顺序不匹配

然而，当我 运行:

df = dd.read_json('*.jsonl', blocksize = 2**29, meta = meta)
keep = ['item1', 'item2', 'item3']
df['x1'] = df[df['x1'].isin(keep)]
df.to_csv(dask_file.csv', single_file = True)

写出了我需要的东西，虽然看进度的时候速度慢了不少。有人可以帮忙吗？

谢谢。

我的猜测是，对于较小的 blocksize，您在应用 .isin(keep) 过滤器后最终得到的分区是空的。如果是这种情况，您可以使用 .

中的函数删除空分区

但是，在这种情况下，我更喜欢 dask.bag API，因为它非常适合与 json 一起使用，请参阅 example。如果你想在最后输出 csv，你可以使用 .to_dataframe().

将包转换为数据帧