问题标签 [dask-distributed]

我需要将非常大的元素提交dask.bag到非线程安全存储，即我需要类似的东西

我不能使用compute，因为袋子太大而无法放入内存。我需要更多类似的东西，distributed.as_completed但它适用于袋子，但distributed.as_completed事实并非如此。

我正在尝试重写一个用类开发的整个项目。渐渐地，最重的计算块应该被并行化，显然我们有很多独立的顺序循环。模仿行为的类的一个例子是这个玩具问题（我是一个痴迷于 p-sum 的数学家）：

如果我将最后一行替换为：

我的运行时间减少了 4（我的机器上可用的内核数）。如果我使用数据密集型的真实类（大熊猫结构占用内存），这种理想行为不会持续存在。是否有 dask.distributed 的推荐替代方案？我看到严重的减速，我将其归因于传递的数据。

我最近将 dask-0.15.3 升级到了 dask-0.16.0 并将distribute-1.19.1 升级到了distribute-1.20.2。升级后，所有 dask 作业都失败并出现异常：_pickle.UnpicklingError: state is not a dictionary

如果我缺少任何图书馆，请告诉我。

这是详细的例外情况：

我正在从事一个涉及一些大于内存的数据集的项目，并且一直在评估用于在集群上而不是在我的本地机器上工作的不同工具。一个看起来特别有趣的项目是 dask，因为它的 DataFrame 类具有与 pandas 非常相似的 API。

我想汇总时间序列相关数据的时间导数。这显然需要按时间戳对时间序列数据进行排序，以便您在行之间取得有意义的差异。但是，dask DataFrames 没有sort_values方法。

在使用 Spark DataFrame 和使用Window函数时，对分区内的排序提供了开箱即用的支持。也就是说，您可以执行以下操作：

然后我可以使用这个窗口函数来计算差异等。

我想知道是否有办法在 dask 中实现类似的东西。原则上，我可以使用 Spark，但我有点时间紧迫，而且我对它的 API 的熟悉程度远不如对 pandas 的熟悉。

在 OSX 10.12.6 上运行 Dask 0.16.0 时，我无法将本地连接dask-worker到本地dask-scheduler。我只是想遵循官方Dask 教程。重现步骤：

第 1 步：运行dask-scheduler

第 2 步：运行dask-worker 10.160.39.103:8786

问题似乎与 dask 调度程序有关，而不是工作人员，因为我什至无法通过其他方式（例如，nc -zv 10.160.39.103 8786）访问端口。

但是，该过程显然仍在机器上运行：

这是对这个问题的跟进。我现在正在尝试在 AWS 上的多个 EC2 节点上运行 Dask。

我能够在第一台机器上启动调度程序：

然后我在其他几台机器上启动工人。从其他机器我可以使用 访问调度程序nc -zv ${HOST} ${PORT}，并且工作人员似乎能够连接到主服务器，正如工作人员的 sysout: 所证明的那样Registered to: tcp://10.201.101.108:31001，但工作人员几乎立即抱怨超时循环。

然后从主节点，在我的 Jupyter 笔记本中连接到调度程序：

dask_client = Client('10.201.101.108:31001')

但是工作不会传播到工作节点（工作节点 CPU 保持在 <1%），甚至不会传播到与调度程序在同一台机器上运行的工作节点。这是一个高度并行化的任务，并且在单台机器上运行时（即使用Client(processes=False)消耗机器上的每个内核）。

我有一个简单的任务，它由 dask-scheduler 调度并在工作节点上运行。

我的要求是，我希望能够在用户想要的时候按需停止任务。

使用 dask-ec2 脚本设置集群后，我尝试将 sql 表读取为 dask 数据帧。我的查询看起来像：

其中 sql_table_name 和 uri 是特定于我的数据库的字符串。然后我尝试使用 dask.distributed 客户端持久化 df：

进度条在 0.6 秒时显示异常。我使用 c.get_futures_error(df) 尝试了解原因并返回如下内容：

我将不胜感激有关为什么我可能会遇到这些期货错误以及我可以做些什么来减轻它们的任何指导。

蟒蛇代码::

运行上述代码后，我收到 proc 对象的序列化错误。请帮我解决这里的问题..

不知道这里出了什么问题..感谢您的时间和帮助..谢谢..

我正在使用 IBM 的 LSF 平台并行运行我的代码。目前，这需要“手动”将代码分解为作业数组；代替：

我分发f了 100 多名工人，然后“手动”将他们所有 100 个不同的结果收集到x.

我试图了解是否dask.distributed可以用作我的模拟内核（在本例中为 IPython 内核）和 LSF 调度程序之间的“桥梁”，以一种以自动方式分发和收集计算的方式。

我找不到有关此的任何文档...任何帮助将不胜感激！