如何在 Dask 中有效地从 DataFrame 转到多个 Series？

Question

我正在尝试找到一种有效的方法将 DataFrame 转换为 Dask 中的一组持久化 Series（列）。

假设数据大小远大于工作器内存的总和，并且大多数操作将被 read-from-disk / spill-to-disk。对于仅对单个列（或列对）进行操作的算法，为每个列操作从磁盘读取整个 DataFrame 是低效的。在这种情况下，最好从本地（可能是持久化的）DataFrame 切换到持久化的列。天真地实施：

persisted_columns = {}
for column in subset_of_columns_to_persist:
    persisted_columns[column] = df[column].persist()

这可行，但效率非常低，因为df[column] 会从磁盘重新读取整个DataFrame N = len(subset_of_columns_to_persist) 次。是否可以基于单个 read-from-disk 反序列化操作单独提取和保留多个列？

---

注意：len(subset_of_columns_to_persist) 是 >> 1，即简单地将 DataFrame 投影到 df[subset_of_columns_to_persist] 不是我正在寻找的解决方案，因为与持久化单个列相比，它仍然具有显着的 I/O 开销。

Answer 1

您可以使用dask.persist 函数同时保存多个集合。这将共享中间体。

columns = [df[column] for column in df.columns]
persisted_columns = dask.persist(*columns)
d = dict(zip(df.columns, persisted_columns))