在 dask 中合并多个大型数据帧

Question

我有多个（~50）大（每个~1到5gb）csv文件，我想将它们合并到一个大的csv文件中。使用 dask，我创建了一个循环，该循环打开每个 csv 并在循环结束时将所有内容保存到 csv 之前调用合并：

for i, file in enumerate(files):

    table = dd.read_csv(file)

    if i == 0:

        merged = table

    else:

        merged = dd.merge(merged, table, on = ["id", "time", "key"])

merged.to_csv("merged\merged1810_*.csv")

但是，这样做会导致MemoryError.

我虽然 Dask 应该分块执行计算以避免这些类型的错误，但我的数据对于 Dask 来说是否太大而无法处理？有没有更好的方法来解决这个问题？

Answer 1

你知道你的文件/数据是如何组织的还是完全随机的？例如，您能否轻松地将您的 csv 文件重新排列为多个文件，例如，每个（ID，密钥）的所有时间？然后你可以按顺序合并每个文件并附加到一个更大的文件？假设每个“分区”都适合内存。如果您没有太多 (id, key) 对，这可能是可以管理的。我遇到了类似的问题，最终做了这样的事情并与 dask 并行。可能不是最有效/最漂亮的，但它完成了工作......希望能帮助到你，