问题标签 [dask-delayed]

如何正确使用延迟的任务来对多列进行分组商数计算？

一些样本数据

一些确定字符串/分类列的设置代码

主要部分：分组商的计算

现在实际上在多个列上调用它

当尝试使用延迟的 dask 时，我需要调用计算来破坏 DAG。为了创建一个并行计算的大型计算图，我该如何解决这个问题？

我想用它Dask来读取一个大型数据集并为其提供Keras模型。数据由音频文件组成，我正在使用自定义函数来读取它们。我试图应用delayed到这个函数，并将所有文件收集在一个 dask 数组中，如下所示：

（见源码）

为了训练 Keras 模型，我如上所述计算 X 和 Y，并将它们输入到函数fit中。

但是，训练非常缓慢。我试图改变chunksize它，它仍然很慢。

你能告诉我在创建数组时我是否做错了什么？或者有什么好的做法？

谢谢

所以基本上我想要的是并行运行 ML Pipelines。我一直在使用 scikit-learn，我决定使用DaskGridSearchCV.

我有一个gridSearchCV = DaskGridSearchCV(pipeline, grid, scoring=evaluator)对象列表，我按顺序运行每个对象：

如果我有 N 个不同的GridSearch对象，我想尽可能多地利用所有可用资源。如果有资源可以同时并行运行 2、3、4、... 或 N，我想这样做。

所以我开始根据 dask 的文档尝试一些事情。首先我尝试了dask.threaded，dask.multiprocessing但它最终变慢了，我不断得到：

/Library/Python/2.7/site-packages/sklearn/externals/joblib/parallel.py:540: UserWarning: Multiprocessing backed parallel loops cannot be nested below threads, setting n_jobs=1

这是代码片段：

也许我正在以错误的方式接近这个，你对我有什么建议吗？

我正在尝试将带有分类列的 Dask 数据框存储到本教程*.h5中的文件中- 1:23:25 - 1:23:45。

这是我对store函数的调用：

功能store是：

并使用分类。

ddf并且stored是类型：

当我运行compute(*[stored])orstored.compute()时，我得到这个：

有没有办法Tag用函数实现列的这种分类store？或者我应该使用不同的方法来存储分类的 Dask 数据帧？

我的目标是分别从多个 numpy 数组中生成多个 png 文件，这些数组是从我的 HD 中的医学图像加载的。为了让事情变得更快，我正在使用 dask 延迟。这是我的工作代码：

我的问题是，每次运行后，一些输出图像都是空的（png 中没有任何内容），并且哪些图像是空的似乎很随机，因为所有输入数据都是有效的。

我怀疑这是多处理和 matplotlib 的问题，如其他相关线程中所见。

有人对如何使用它有建议dask吗？

编辑：最小的工作示例

与这个问题类似，我遇到了分布式 Dask 的内存问题。但是，在我的情况下，解释并不是客户试图收集大量数据。

这个问题可以基于一个非常简单的任务图来说明：一个delayed操作列表生成一些固定大小约为 500 MB 的随机数据帧（以模拟从文件加载许多分区）。任务图中的下一个操作是获取每个 DataFrame 的大小。最后，所有大小都被缩减为一个总大小，即必须返回给客户端的数据很小。

出于测试目的，我正在运行一个本地调度程序/工作程序单线程，限制为 2GB 内存，即：

我对任务图的期望是，worker 永远不需要超过 500 MB 的 RAM，因为在“生成数据”之后直接运行“获取数据大小”应该会使数据立即变小。但是，我观察到工作人员需要比这更多的内存：

因子 2 表示必须在内部复制数据。因此，任何使分区大小接近节点物理内存的尝试都会导致MemoryErrors或大量交换。

任何可以阐明这一点的信息都将受到高度赞赏。尤其：

• 我是否可以控制数据的重复，这是可以避免的吗？还是一般的经验法则是将有效负载保持在远低于 50% 以解决数据重复问题？
• 工人如何memory-limit影响这种行为？从我的测试来看，使用较低的阈值似乎会更早地触发 GC（和/或溢出到磁盘？），但另一方面，还有其他内存峰值甚至超过了使用更高阈值的峰值内存。

请注意，我知道我可以通过在第一次操作中获取大小来解决这个特定问题，并且可能 Dask 的单机执行器更适合这个问题，但我要求的是教育目的。

附件一：测试代码

附件 2： DAG 插图

我正在构建一个非常大的 DAG 以提交给分布式调度程序，其中节点对本身可能非常大的数据帧进行操作。一种模式是我有大约 50-60 个函数来加载数据并构建每个数百 MB 的 pandas 数据帧（并在逻辑上表示单个表的分区）。我想将这些连接到图中下游节点的单个 dask 数据帧中，同时最大限度地减少数据移动。我像这样链接任务：

在哪里

我已经尝试了各种concat_all实施，但这似乎是合理的：

所有 pandas 数据框的索引都是不相交的，并且是排序/单调的。

concat_all但是，即使每个人的内存预算实际上都相当大，我也不希望它会移动数据，但我正在杀死死于此功能的工作人员（集群管理器正在杀死他们超过他们的内存预算）。我有理由确定，compute()在使用 dask 数据帧的图形节点中调用之前，我总是切分到合理的数据子集。

--memory-limit到目前为止，我正在玩但没有成功。我至少正确地解决了这个问题吗？有没有我遗漏的注意事项？

我是尝试延迟延迟的新用户。我想使用延迟将函数和代码自动转换为延迟。但是，我发现delayed.compute 没有递归地计算收集中的延迟......

我是否遗漏了任何使其工作的东西或 Dask.delayed 不支持它？

在使用延迟转换我的程序时，我偶然发现了一种不适用于延迟的常用编程模式。例子：

提高：TypeError: Delayed objects of unspecified length are not iterable 虽然以下解决方法没有。但是看起来笨拙了很多

这是预期的行为吗？

有没有更快的方法来只用 Dask 检索大型已发布数组中的单个元素而不检索整个数组？

在下面的示例中，client.get_dataset('array1')[0] 与 client.get_dataset('array1') 所用的时间大致相同。