集群上数据的 Dask 和持久化答案

【问题标题】：Dask and persistence of data on the cluster集群上数据的 Dask 和持久化
【发布时间】：2022-11-11 01:20:47
【问题描述】：

我正在研究一个使用历史数据和传入数据进行分析的项目。我想学习如何管理更新 dask 上的传入数据，而不必每次都发送所有历史数据。

我为时间序列收集数据进行分析，但时间序列随着传入数据的增长而增长，并且每个流的传入数据需要发送给适当的工作人员进行 ARMA 分析等工作。如果我对天气进行 ARMA 分析，我希望将气压与温度分开，并通过比较压力与压力以及温度与温度来进行分析。我不想将新的温度数据附加到先前的温度数据中，并将现在更大的系列发送给新的工人。我只想将新的温度数据发送给已经拥有所有先前温度数据的 dask 工作人员，依此类推。如何确保先前的温度数据保留在工作人员身上，以及如何（仅）将新的温度数据发送给拥有先前数据的工作人员。

我已经用 dask 做了一些基本的事情，但是所有的基本课程都没有解决历史和方法关于结果的工人唯一持久性。

此外，这些数据不是基于 Dask 系列或数据帧，而是基于包含与分析方法相关的不同数据和方法的类。所以我不能有效地使用 dask 系列或数据框。

任何帮助，将不胜感激

【问题讨论】：

而不是一个比喻，一个更具体的例子会有所帮助。你想做什么？
@MichaelDelgado我更新了解释以更接近确切的问题而不是隐喻。
这是 dask 的一个相对不常见的用例，但它肯定是可能的。只是要明确一点-您的想法是永远运行 dask 作业，使用在内存中维护数据并等待响应请求的工作人员？通常，我建议您使用纯函数或至少最终一致的函数来提高可靠性。但如果你愿意，你可以建立这样一个模型。要检查的一件事是dask.distributed Actors，它允许有状态和面向对象的工作流。不过，我肯定会阅读有关限制的部分。
一般来说，如果没有更多细节，很难具体回答这个问题。如果你想要一个示例实现，纯 python 中的minimal reproducible example 我们可以帮助迁移到 dask 将是一个有用的起点。

标签： python dask directed-acyclic-graphs

【解决方案1】：

这可能不是正确的解决方案，但一种可能性是指定特定的工作人员来执行特定的计算。例如，让我们将工人分成两组：

# instantiate workers
from distributed import Client
c = Client(n_workers=5)

# here the separation is done based on order
# but custom logic can be implemented instead
workers_pressure = list(c.scheduler_info()['workers'])[3:]
workers_temperature = list(c.scheduler_info()['workers'])[:3]

现在，对于与pressure 相关的任务，我们可以指定与pressure 相关的工作人员：

data_pressure = [4,5,6]
data_temperature = [1,2,3]

# scatter data to pressure/temperature workers
d_p = client.scatter(data_pressure, workers= workers_pressure)
d_t = client.scatter(data_temperature, workers=workers_temperature)

# submit computations to specific workers
function_pressure = lambda x: x**2
function_temperature = lambda x: x**2

f_p = client.map(function_pressure, d_p, workers=workers_pressure)
f_t = client.map(function_temperature, d_t, workers= workers_temperature)

在上面的 sn-p 中，指定处理压力数据的工作人员将用于运行压力计算。

如果您有一组非常异构的任务，这将无法很好地扩展。如果这是您的情况，我将首先构建任务图 (DAG)，然后让dask 处理最有效的工作人员分配任务。

【讨论】：