问题标签 [dask-distributed]

我正在将多个操作应用于 dask 数据帧。我可以为特定操作定义分布式工作人员资源要求吗？

例如，我称之为：

我想为map_partitions()指定资源要求（可能与map()的资源要求不同），但似乎该方法不接受资源参数。

PS。或者，我发现我可以在map_partitions( )之后调用client.persist( )并在此调用中指定资源，但这会立即触发计算。

我想知道在使用 Dask 进行 groupBy 聚合之后，是否可以从给定列中获取唯一项的数量。我在文档中看不到这样的东西。它在 pandas 数据框上可用并且非常有用。我已经看到了一些与此相关的问题，但我不确定它是否已实施。

有人可以给我一些提示吗？

我正在尝试使用 dask(async) 框架执行一个简单的任务（一个实例方法），但它因序列化错误而失败。

有人可以指出我正确的方向。

这是我正在运行的代码：

错误：

Distributed.protocol.pickle - 信息 - 无法在 0x101c408d0>> 处序列化 main.TestDask 对象。例外：无法腌制 select.kqueue 对象 Traceback（最近一次调用最后一次）：

在 Kubernetes 和 AWS 上测试自动缩放 Dask 分布式实现时，我创建了一个演示问题，但我不确定我是否正确解决了这个问题。

我的场景是给定一个字符串的 md5 哈希（代表密码）找到原始字符串。我遇到了三个主要问题。

A) 参数空间很大，尝试创建一个包含 2.8211099e+12 个成员的 dask bag 会导致内存问题（因此您将在下面的示例代码中看到“explode”函数）。

B) 早期发现时干净的出口。我认为 usingtake(1, npartitions=-1)会实现这一点，但我不确定。最初我提出了一个raise Exception("%s is your answer' % test_str)有效但感觉“肮脏”的异常

C) 鉴于这是长期运行，有时工作人员或 AWS 盒子会死机，如何最好地存储进度？

示例代码：

我发现“简单”在本地运行良好，“中等难度”在我们的 6 到 8 * m4.2xlarge AWS 集群上运行良好。但到目前为止还没有hard工作。

I have launched a dask.distributed cluster via slurm (using dask-mpi) across a number of cores on a slurm-managed cluster. All the processes appear to have started OK (normal-looking stdout in the slurm logfile), but when I try to connect a client from within python using client = Client(scheduler_file='/path/to/my/scheduler.json'), I get a timeout error as follows:

These are the contents of scheduler.json after launch. I don't know if not listing the worker processes here is normal or not, or if this signifies some problem with setup:

I have gotten the same issue on two different slurm-managed clusters. Does it look like I need to specify something specific for the ports or something? If so, how do I go about figuring out what ports I need to use?

作为数据工作流的一部分，我需要修改 dask 数据框列子集中的值，并将结果传递给进一步计算。特别是，我对 2 种情况感兴趣：映射列和映射分区。推荐的安全和高效的数据处理方式是什么？我在每个主机上有多个工作进程的集群上运行它的分布式设置。

情况1。

我想跑：

这会返回一个数据系列，所以我假设原始数据框没有被修改。将列分配回数据框是否安全，例如dataframe['column']=res？可能不是。我是否应该使用.copy()进行复制，然后将结果分配给它，例如：

还有其他推荐的方法吗？

案例2

我需要映射数据框的分区：

在mapping_func()内部，我想修改所选列中的值，方法是使用partition[column].map或简单地创建一个列表理解。同样，如何安全地修改分区并从映射函数中返回它？

映射函数接收的分区是 Pandas 数据帧（原始数据的副本？），但是在就地修改数据时，我看到了一些崩溃（虽然没有异常/错误消息）。打电话也是一样partition.copy(deep=False)，没用。分区是否应该被深度复制然后就地修改？或者我应该总是从新的/映射的列数据和原始/未修改的系列/列中构建一个新的数据框？

从 jupyter notebook 使用 dask 时，我看到了奇怪的行为。所以我正在启动一个本地客户，并给它一份工作清单。我的真实代码有点复杂，所以我在这里为你举一个简单的例子：

问题是，我意识到： 1. Dask 在这个例子中启动了超过 9 个进程。2. 代码运行完成后（notebook 中没有任何东西在运行），dask 创建的进程不会被杀死（客户端也不会关闭）。当我做一个顶部时，我可以看到所有这些过程仍然存在。

我在文档中看到有一个 client.close() 选项，但有趣的是，0.15.2 中不存在这样的功能。

dask 进程被杀死的唯一时间是我停止 jupyter notebook 的时候。此问题导致奇怪且不可预测的性能行为。当笔记本上没有运行代码时，进程是否会被终止或客户端关闭？

语境

我正在使用 dask.distributed 跨机器并行计算。因此，我有 dask-workers 在连接到 dask-scheduler 的不同机器上运行，然后我可以将我的自定义图表与所需的键一起提交到该机器上。

由于网络安装限制，我的输入数据（和输出存储）仅可用于机器的子集（'i/o-hosts'）。我试图以两种方式处理这个问题：

1. i/o 操作中涉及的所有任务都限制在 i/o 主机上（它们只能在运行在可以访问数据的机器上的工作人员上运行），非 i/o 任务仅限于非 i/o 主机上
2. i/o 操作中涉及的所有任务都绑定到提供资源“io”（i/o 主机）的工作人员，非 i/o 任务绑定到提供资源“计算”的非 i/o 主机上的工作人员'。

不允许非 i/o 任务在 i/o 主机上运行的想法是确保他们的工作人员可用于 i/o 任务。

问题

这两种方法都按预期工作，因为它们将 i/o 任务限制在正确的工作人员身上。但是我注意到，当使用这两种方法中的任何一种时，只有极少数的工作人员会积累大量的任务，而其他大多数工作人员仍然处于空闲状态。

在阅读了如何在工作人员之间分配任务后，我发现工作窃取似乎是故意禁用的受限任务（http://distributed.readthedocs.io/en/latest/work-stealing.html）。这似乎也适用于资源框架。

问题

有没有一种好方法可以将任务限制与窃取工作结合起来？

我的目标是在一组 3 个 linux (ubuntu) 物理工作站上建立一个 docker swarm 并在其上运行一个dask集群。

我能够初始化 docker swarm 并将所有机器添加到 swarm 中。

然后我docker stack deploy -c docker-compose.yml dask-cluster在领导者框上运行。

这里是docker-compose.yml：

这是richardbrks/dask-cluster：

当我部署 swarm 时，dworker不在同一台机器上的节点dscheduler 不知道是什么dscheduler。我 ssh 进入这些节点之一并查看了 env，dscheduler但并不存在。我也尝试 ping dscheduler，并得到“ping：未知主机”。

我认为 docker 应该为服务发现提供一个基于内部 dns 的服务，这样调用dscheduler就会把我带到dschedler节点的地址。

我的电脑是否有一些我遗漏的设置？还是我的任何文件丢失了什么？

所有这些代码也位于https://github.com/MentalMasochist/dask-swarm

我在本地使用 dask 客户端。问题是当任务完成时，即使我使用本地调度程序，我仍然在内存中有很多 python 实例。这就是我正在做的事情：

如果我多次运行它，我最终会在我的进程中（在 Windows 上）有几个 python.exe。有什么办法可以防止这种情况？

谢谢