问题标签 [dask-distributed]

我的dask groupby脚本失败（Memory Error），所以我开始调试脚本。我在单机电脑上运行。
我已经更新logging了config.yaml文件中的
logging: distributed: debug bokeh: debug tornado: info

我正在运行具有以下更新 的dask 分布式示例计算：

和一个更新的脚本，range修改为range(10000).

在脚本运行（并成功完成）时，我会在其中看到以下日志， 这些日志会不断Jupyter Notebook重复，直到脚本完成。

bokeh.server.tornado - 调试 - [pid 3088] 0 个客户端已连接
bokeh.server.tornado - 调试 - [pid 3088] /system 有 0 个会话，其中 0 个未使用 bokeh.server.tornado - 调试 - [pid 3088] /stealing 有0 个会话，0 个未使用的 bokeh.server.tornado - 调试 - [pid 3088] /workers 有 0 个会话，0 个未使用的 bokeh.server.tornado - 调试 - [pid 3088] /events 有 0 个会话，0 个未使用的 bokeh.server.tornado - 调试 - [pid 3088] /counters 有 0 个会话，其中 0 个未使用的 bokeh.server.tornado - 调试 - [pid 3088] /tasks 有 0 个会话，其中 0 个未使用的 bokeh.server.tornado - 调试 - [pid 3088] /status 有0 个会话，0 个未使用

为什么我看不到任何正在运行的工人？

运行时我得到以下日志c = LocalCluster()

bokeh.server.server - 信息 - 启动 Bokeh 服务器版本 0.12.4
bokeh.server.server - 警告 - 主机通配符“ ”可能会将应用程序暴露给 HTTP 主机标头攻击。主机通配符只能用于测试目的。bokeh.server.server - 警告 - 主机通配符“ ”可以将应用程序暴露给 HTTP 主机标头攻击。主机通配符只能用于测试目的。bokeh.server.tornado - 调试 - 允许的主机头：[' '] bokeh.server.tornado - 调试 - 这些主机来源可以连接到 websocket：[' '] bokeh.server.tornado - 调试 - 模式是：散景。 server.tornado - DEBUG - [('/system/?', bokeh.server.tornado - DEBUG - , bokeh.server.tornado - DEBUG - {'application_context': , bokeh.
'bokeh_websocket_path': '/system/ws'}), bokeh.server.tornado - DEBUG - ('/system/ws', bokeh.server.tornado - DEBUG - , bokeh.server.tornado - DEBUG -
{'application_context' : , bokeh.server.tornado - 调试 -
'bokeh_websocket_path': '/system/ws'}), bokeh.server.tornado - 调试 - ('/system/autoload.js', bokeh.server.tornado - 调试 - , bokeh.server.tornado - DEBUG - {'application_context': , bokeh.server.tornado - DEBUG -
'bokeh_websocket_path': '/system/ws'}), bokeh.server.tornado - DEBUG - ('/stealing/?' , bokeh.server.tornado - 调试 - , bokeh.server.tornado - 调试 - {'application_context': , bokeh.server.tornado - 调试 -
'bokeh_websocket_path': '/stealing/ws'}), bokeh.server.tornado - DEBUG - ('/stealing/ws', bokeh.server.tornado - DEBUG - , bokeh.server.tornado - DEBUG -
{'application_context' : , bokeh.server.tornado - 调试 -
'bokeh_websocket_path': '/stealing/ws'}), bokeh.server.tornado - 调试 - ('/stealing/autoload.js', bokeh.server.tornado - 调试 - , bokeh.server.tornado - DEBUG - {'application_context': , bokeh.server.tornado - DEBUG -
'bokeh_websocket_path': '/stealing/ws'}), bokeh.server.tornado - DEBUG - ('/workers/?' , bokeh.server.tornado - 调试 - , bokeh.server.tornado - 调试 - {'application_context': , bokeh.server.tornado - 调试 -
'bokeh_websocket_path': '/workers/ws'}), bokeh.server.tornado - DEBUG - ('/workers/ws', bokeh.server.tornado - DEBUG - , bokeh.server.tornado - DEBUG -
{'application_context' : , bokeh.server.tornado - 调试 -
'bokeh_websocket_path': '/workers/ws'}), bokeh.server.tornado - 调试 - ('/workers/autoload.js', bokeh.server.tornado - 调试 - , bokeh.server.tornado - DEBUG - {'application_context': , bokeh.server.tornado - DEBUG -
'bokeh_websocket_path': '/workers/ws'}), bokeh.server.tornado - DEBUG - ('/events/?' , bokeh.server.tornado - 调试 - , bokeh.server.tornado - 调试 - {'application_context': , bokeh.server.tornado - 调试 -
'bokeh_websocket_path': '/events/ws'}), bokeh.server.tornado - DEBUG - ('/events/ws', bokeh.server.tornado - DEBUG - , bokeh.server.tornado - DEBUG -
{'application_context' : , bokeh.server.tornado - 调试 -
'bokeh_websocket_path': '/events/ws'}), bokeh.server.tornado - 调试 - ('/events/autoload.js', bokeh.server.tornado - 调试 - , bokeh.server.tornado - DEBUG - {'application_context': , bokeh.server.tornado - DEBUG -
'bokeh_websocket_path': '/events/ws'}), bokeh.server.tornado - DEBUG - ('/counters/?' , bokeh.server.tornado - 调试 - , bokeh.server.tornado - 调试 - {'application_context': , bokeh.server.tornado - 调试 -
'bokeh_websocket_path': '/counters/ws'}), bokeh.server.tornado - DEBUG - ('/counters/ws', bokeh.server.tornado - DEBUG - , bokeh.server.tornado - DEBUG -
{'application_context' : , bokeh.server.tornado - 调试 -
'bokeh_websocket_path': '/counters/ws'}), bokeh.server.tornado - 调试 - ('/counters/autoload.js', bokeh.server.tornado - 调试 - , bokeh.server.tornado - DEBUG - {'application_context': , bokeh.server.tornado - DEBUG -
'bokeh_websocket_path': '/counters/ws'}), bokeh.server.tornado - DEBUG - ('/tasks/?' , bokeh.server.tornado - 调试 - , bokeh.server.tornado - 调试 - {'application_context': , bokeh.server.tornado - 调试 -
'bokeh_websocket_path': '/tasks/ws'}), bokeh.server.tornado - DEBUG - ('/tasks/ws', bokeh.server.tornado - DEBUG - , bokeh.server.tornado - DEBUG -
{'application_context' : , bokeh.server.tornado - 调试 -
'bokeh_websocket_path': '/tasks/ws'}), bokeh.server.tornado - 调试 - ('/tasks/autoload.js', bokeh.server.tornado - 调试 - , bokeh.server.tornado - DEBUG - {'application_context': , bokeh.server.tornado - DEBUG -
'bokeh_websocket_path': '/tasks/ws'}), bokeh.server.tornado - DEBUG - ('/status/?' , bokeh.server.tornado - 调试 - , bokeh.server.tornado - 调试 - {'application_context': , bokeh.server.tornado - 调试 -
'bokeh_websocket_path': '/status/ws'}), bokeh.server.tornado - DEBUG - ('/status/ws', bokeh.server.tornado - DEBUG - , bokeh.server.tornado - DEBUG -
{'application_context' : , bokeh.server.tornado - 调试 -
'bokeh_websocket_path': '/status/ws'}), bokeh.server.tornado - 调试 - ('/status/autoload.js', bokeh.server.tornado - 调试 - , bokeh.server.tornado - DEBUG - {'application_context': , bokeh.server.tornado - DEBUG -
'bokeh_websocket_path': '/status/ws'}), bokeh.server.tornado - DEBUG - ('/?', bokeh .server.tornado - DEBUG - , bokeh.server.tornado - DEBUG - {'applications': {'/counters': , bokeh.server.tornado - DEBUG -
'/events': , bokeh.server.tornado - 调试 -
'/status': , bokeh.server.tornado - 调试 -
'/stealing': , bokeh.server.tornado - 调试 -
'/system': , bokeh.server.tornado - 调试 -
'/tasks': , bokeh.server.tornado - 调试 -
'/workers': }, bokeh .server.tornado - DEBUG -
'prefix': '', bokeh.server.tornado - DEBUG - 'use_redirect': True}), bokeh.server.tornado - DEBUG - ('/static/(.*)', bokeh .server.tornado - 调试 - )]

我想提交一个 dask 任务，它将执行以下操作：

1. def fakejob使用 dask.bag ( )构建惰性 dask 图
2. 从 1. 计算图形并将其保存到镶木地板（这部分省略，只是一个动机）

我需要对多个输入执行此操作，因此我一直在尝试像这样使用 dask.distributed 的期货功能。

问题是我不断得到：AssertionError: daemonic processes are not allowed to have children

我已经尝试关注这个链接并最终得到了第二个版本（与第一个版本不同的是 1 行），但期货永远保持“待定”。

关于如何做到这一点的任何线索？

干杯。

我一直在阅读有关从工作人员上运行的任务中生成新进程的文档。我从这里遇到了这个：

但是，每个正在运行的任务都占用一个线程，因此如果您启动了许多启动其他任务的任务，那么如果您不小心，可能会导致系统死锁。您可以从任务中调用 secede 函数，以将其从专用线程池中移除到不占用 Dask worker 中的插槽的管理线程中

转移到管理线程是什么意思？是否所有插槽都具有相同的优先级或类似的东西？并且管理线程确实有优先权？

作为一个具体的例子，这是我想尝试的：

wherecompute_square和该lambda函数可能会运行很多次，并且将是一个计算量更大的函数，值得提交作业的 1ms 开销。我应该secede在这种情况下使用吗？

我对 dask 文档中的并发期货要点有疑问：https ://gist.github.com/mrocklin/ef9ccd29a6ec5f4de84d6192be95042a

当我们实例化一个 Executor 时，这是否意味着它也实例化了一个调度器？执行人的代表说

<Client: scheduler='tcp://127.0.0.1:60329' processes=4 cores=4>

那么调用Client()和Executor()有什么区别呢？如果不一样，文档似乎非常相似。在并发期货示例中，一些首先实例化一个客户端，然后使用客户端的 IP 地址实例化一个执行器。

这是我之前关于使用 Dask 计算访问大数组中的一个元素的问题之一的潜在答案的后续问题。

为什么使用 Dask 计算会导致执行在下面挂起？这是工作代码片段：

我无法使用读取的数据帧访问我在 S3 上的文件：df_read_csv。我得到错误：Exception: Unable to locate credentials

当我的分布式 dask 针对本地工作核心运行时，这可以正常工作。但是，当我导入带有附加工作服务器集群的客户端时，它会失败。我的集群是在 ubuntu 上使用 dask-ec2 创建的，头服务器上有 1 个调度程序和 3 个工作服务器（都是 ubuntu）。

我假设失败是因为所有工作人员也需要访问 S3。我已经在所有这些设备上安装了 aws cli 并使用我的密钥进行连接，并且可以从 cli 中列出 S3 存储桶。但是，由于某种原因，我的数据框读取引发了一个 ubuntu 错误，指出 boto 无法找到凭据

我浏览了各种帖子，但似乎找不到任何有用的东西。这是错误的屏幕截图：

dask.compute(...) 应该是一个阻塞调用。但是，当我嵌套了 dask.compute，而内部的执行 I/O（如 dask.dataframe.read_parquet）时，内部的 dask.compute 不会阻塞。这是一个伪代码示例：

如果我启动 2 个工作人员，每个工作人员有 8 个进程，例如：

，那么我预计最多 2 x 8 个并发的 inner_func 正在运行，因为 inner_func(files).compute() 应该是阻塞的。然而，我观察到的是，在一个工作进程中，一旦它开始 read_parquet 步骤，可能会有另一个 inner_func(files).compute() 开始运行。所以最后可能会有多个 inner_func(files).compute() 运行，有时它可能会导致内存不足错误。

这是预期的行为吗？如果是这样，有什么方法可以强制每个工作进程执行一个 inner_func(files).compute() 吗？

在 dask.distributed 上使用期货时，有没有办法区分pending当前正在评估的期货和仍在队列中的期货？

原因是我将大量任务（~8000）提交给一小部分工作人员（100），因此并非所有任务都可以立即处理。这些任务涉及调用第三方可执行文件 ( via subprocess.check_output )，在极少数情况下会进入无限循环。

因此，我想取消运行时间过长的期货（使用任意超时）。然而，似乎没有办法判断未来是否已经处于一种pending状态很长时间，因为计算时间比平时长，或者仅仅是因为它必须等待工作人员可用。

我的设置涉及一个 SGE 集群，分别运行 adask-scheduler和dask-workerjob/job-array。@timeout_decorator.timeout(60, use_signals=False)我尝试使用timeout_decorator 包直接在提交的 Python 函数中设置超时，但出现以下错误：

任何帮助将非常感激。

我是新手 do dask（作为 dd 导入）并尝试转换一些 pandas（作为 pd 导入）代码。

以下几行的目标是将数据切片到这些列，这些值满足 dask 中计算的要求。

csv中有一个给定的表。以前的代码读取

并且工作正常。我对转换的简单想法是将第一行替换为

但这导致了奇怪的错误消息IndexError: too many indices for array。即使切换到准备好的计算数据，inputdata错误pseudob仍然存​​在。
也许这个问题专门针对 dask-columns 计算布尔切片的想法。

我刚刚找到了一种（可能不是最佳的）方法（不是解决方案）来做到这一点。将第 4 行更改为以下内容

似乎工作。

我在 Linux CentOS 7 上运行 Dask Distributed，安装了 Python 3.6.2。我的计算似乎越来越好（我仍在改进我的代码，但我能够得到一些结果），但我不断收到一些显然与 tornado 模块相关的 python 错误。我只启动一个单节点独立 Dask 分布式集群。

这是最常见的例子：

这是另一个：

你知道为什么吗？我应该关闭一些连接还是停止独立集群？