问题标签 [dask-distributed]

如果在短时间内提交大量任务，我遇到了一个问题，即 dask 调度程序被杀死（尽管工作人员继续运行）并出现内存错误。

如果可以获取集群上当前的任务数，那么就很容易控制提交到集群的并发任务数。

注意：任务正在从多个客户端提交到同一个调度程序。

使用 fire_and_forget 提交任务图时，是否可以稍后（在新进程/客户端中）取消这些任务（例如通过键名）？这也会取消所有相关的任务，还是这些也像“fire_and_forget”一样？

我正在使用 dask.distributed 集群，我想将一个大数据帧保存到一个 CSV 文件到 S3，如果可能的话保持分区的顺序（默认情况下 to_csv() 将数据帧写入多个文件，每个分区一个）。此外，这个文件存储操作也应该作为惰性/延迟任务执行（它应该在集群工作人员上执行）。

我能想到的第一个解决方案是将分区存储在临时 S3 位置，然后在延迟函数中使用 S3 分段上传将这些文件合并/上传在一起。临时存储在这里是一个缺点。

可以使用 dataframe.to_delayed() 来完成吗？上传需要由单个延迟函数执行（S3没有追加操作），同时dataframe可能大于worker的内存，因此该函数不能简单地依赖所有dataframe分区。

PS。CSV 格式是此用例的要求。

我正在使用dask.distributed调度程序和工作人员来处理集群上的一些大型显微镜图像。我每个节点运行多个工作人员（1 个核心 = 1 个工作人员）。节点中的每个核心共享 200Gb 的 RAM。

问题
我想将写入实例减少到集群的共享 HD。

问题

1. 这个想法是创建一个在节点内共享的字典，用处理过的图像填充它，直到大小达到 RAM 的 80% 左右，然后字典中的每个图像将作为单独的文件保存到 HD。是否可以在节点中的工作人员之间共享字典？
2. 字典中的每个图像都将写入不同的文件。将它们编写为遍历字典或 i/o 调用的速度/和数量将与在处理期间同时编写一张图像相同吗？

我没有当前正在运行的示例，因为我不知道如何在同一节点上的工作人员之间共享变量。

谢谢

这是我用来连接到 hdfs 并创建 dask 数据帧的代码。

错误：

HADOOP 版本：2.5.1

经过几个阶段的惰性数据帧处理后，我需要在保存数据帧之前对其进行重新分区。但是，该.repartition()方法需要我知道分区的数量（而不是分区的大小），这取决于处理后数据的大小，这仍然是未知的。

我想我可以懒惰地计算大小，df.memory_usage().sum()但repartition()似乎不接受它（标量）作为参数。

有没有办法进行这种自适应（基于数据大小）的延迟重新分区？

PS。由于这是我管道中的（几乎）最后一步，我可以通过“手动”转换为延迟和重新分区来解决这个问题（我不需要回到数据框），但我正在寻找一种更简单的方法去做这个。

PS。按分区大小重新分区也是一个非常有用的功能

如何删除已成为孤儿的 Dask 工作人员？我已经尝试过通常的方法，例如 pkill 和 kill -9 但工人不会死。

TL;DR
我想在启动时将数据集预加载到 Dask 分布式调度程序中。

背景
我以实时查询方式使用 Dask，数据集更小。因为它是实时的，所以工作人员可以相信调度程序总是有某些可用的数据集——即使在启动后立即可用，这一点很重要。工作人员始终将整个数据集保存在内存中。

传统上，我通过连接客户端、分散 df 并发布数据集来完成此操作：

但这留下了调度程序重新启动和数据集未加载的可能性。

我知道您可以使用--preload它在启动时执行脚本，如下所示：

样板代码如下所示：

但是如何说服调度程序在不使用外部客户端的情况下加载我的数据集？

从理论上讲，我可能会放弃一个启动预填充客户端的子进程，但感觉不太理想:)

调度程序启动中的普通客户端
尝试在调度程序启动中作为客户端连接：

挂在上面，c = Client(scheduler.address)必须被强行杀死（kill -9）

我正在尝试从 HDFS 文件（csv）创建 dask 数据帧。HDFS 中存储的 csv 文件有很多部分文件。

在 read_csv api 调用中：

出现以下错误：

实际上 /data.csv 是包含许多零件文件的目录。我不确定是否有一些不同的 API 可以读取此类 hdfs csv。

TL;DR
如果在另一个查询正在下载所需数据集时出现多个查询 - Dask 会尝试多次下载数据集吗？或者它会承认它“在飞行中”并自动等待它完成？

背景
如果我有一个刚刚启动的工作人员（尚未将数据集加载到内存中）并且我的函数要求数据集，它将根据需要下载到工作人员上。一个简单的场景：

但是，如果我有以下情况：

Dask 会尝试多次下载数据集，还是会承认它“正在运行”并自动等待它完成？

我已经阅读了源代码，但数据集发布/列表对我来说仍然是一个黑匣子。