python - 在 Dask 中重用中间结果(混合延迟和 dask.dataframe)

根据我在 an earlier question 上收到的答案，我编写了一个 ETL 程序，如下所示:

import pandas as pd
from dask import delayed
from dask import dataframe as dd

def preprocess_files(filename):
    """Reads file, collects metadata and identifies lines not containing data.
    """
    ...
    return filename, metadata, skiprows

def load_file(filename, skiprows):
    """Loads the file into a pandas dataframe, skipping lines not containing data."""
    return df

def process_errors(filename, skiplines):
    """Calculates error metrics based on the information
    collected in the pre-processing step
    """
    ...

def process_metadata(filename, metadata):
    """Analyses metadata collected in the pre-processing step."""
    ...

values = [delayed(preprocess_files)(fn) for fn in file_names]
filenames = [value[0] for value in values]
metadata = [value[1] for value in values]
skiprows = [value[2] for value in values]

error_results = [delayed(process_errors)(arg[0], arg[1])
                 for arg in zip(filenames, skiprows)]
meta_results = [delayed(process_metadata)(arg[0], arg[1])
                for arg in zip(filenames, metadata)]

dfs = [delayed(load_file)(arg[0], arg[1])
       for arg in zip(filenames, skiprows)]
... # several delayed transformations defined on individual dataframes

# finally: categorize several dataframe columns and write them to HDF5
dfs = dd.from_delayed(dfs, meta=metaframe)
dfs.categorize(columns=[...])  # I would like to delay this
dfs.to_hdf(hdf_file_name, '/data',...)  # I would also like to delay this

all_operations = error_results + meta_results # + delayed operations on dask dataframe
# trigger all computation at once,
# allow re-using of data collected in the pre-processing step.
dask.compute(*all_operations)

ETL 过程经过几个步骤:

预处理文件，识别不包含任何相关数据的行并解析元数据

使用收集的信息，处理错误信息、元数据并将数据行并行加载到 Pandas 数据帧中(重新使用预处理步骤的结果)。操作( process_metadata 、 process_errors 、 load_file )具有共享数据依赖性，因为它们都使用在预处理步骤中收集的信息。理想情况下，预处理步骤只运行一次，结果跨进程共享。

最终，将 Pandas 数据帧收集到一个 dask 数据帧中，对它们进行分类并将它们写入 hdf。

我遇到的问题是， categorize 和 to_hdf 会立即触发计算，丢弃元数据和错误数据，否则 process_errors 和 process_metadata 将进一步处理这些元数据和错误数据。

有人告诉我，在 dask.dataframes 上延迟操作会导致问题，这就是为什么我很想知道是否有可能触发整个计算(处理元数据、处理错误、加载数据帧、转换数据帧并以 HDF 格式存储它们) )，允许不同的进程共享在预处理阶段收集的数据。

最佳答案

有两种方法可以解决您的问题:

延迟一切

分阶段计算

延迟一切

to_hdf 调用接受 compute= 关键字参数，您可以将其设置为 False。如果为 False，它将返回一个 dask.delayed 值，您可以随时计算该值。

但是，如果您想继续使用 dask.dataframe，则需要立即计算分类调用。如果不立即或多或少地遍历数据，我们将无法创建一致的 dask.dataframe。最近 Pandas 围绕联合分类的改进将使我们在 future 改变这一点，但现在你被困住了。如果这对您来说是一个障碍，那么您将不得不切换到 dask.delayed 并使用 df.to_delayed() 手动处理一些事情

分阶段计算

如果您使用 distributed scheduler ，则可以使用 .persist method 暂存计算。

from dask.distributed import Executor
e = Executor()  # make a local "cluster" on your laptop

delayed_values = e.persist(*delayed_values)

... define further computations on delayed values ...

results = dask.compute(results)  # compute as normal

这将让您触发一些计算，并且仍然让您继续定义您的计算。您保留的值将保留在内存中。

关于python - 在 Dask 中重用中间结果(混合延迟和 dask.dataframe)，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/39411407/