我有以下项目树

在我的开头，我main.py像这样从 `dask_test.helpers.datetime' 导入一个对象

在我的主要使用下来。在我的主文件中，我定义了一些函数并将它们应用于像这样的 dask Dataframe

转换函数采用 dask.dataframe 并通过使用 apply 他们向其中添加一个新列，如下所示：

尝试使用 dask 分布式和 LocalCluster 进行计算可以正常工作：

但是当打开 dask-scheduler 和 dask-workers 我收到以下消息

return pickle.loads(x) ImportError: No module named dask_test.helpers.datetime

1. 第一个问题 LocalCluster 不使用酸洗？
2. 所有模块都需要包含可腌制对象才能与正确分发的 dask 一起工作？

编辑：

导入 datetime 模块和 cloudpickle 似乎 datetime 是可选的

编辑：经过更多调查后，我在日志文件中看到了这一点：

数据框

在我的主文件中，我首先创建了 pandas Dataframe，然后使用from_pandas方法将它变成了一个 dask DataFrame。

编辑 3：我发现了问题所在，但我不明白为什么。在我的 datetime 模块中，我定义了一个名为 TimeWindow 的对象来处理时间段。我的 datajson 文件有一个带有时间戳的字段，其形式为

我在数据框上应用了一个函数来更改添加一个将上述内容转换为时间窗口对象的列，如下所示：

并在数据框上（这是一个熊猫数据框。我在创建 dask 数据框之前这样做）

没有它，工人工作得很好。但是 TimeWindow 对象和实例是可腌制的

设置基本上是：

客户端：

工人（工人 - 客户端）端：

client.persist我可以通过使用and得到一些工作client.publish_dataset。问题是我正在进行手动跟踪以确定何时执行unpublish_dataset，并且使用异步行为这变得很困难。

现在有（更）自然的方法可以做到这一点吗？

编辑：似乎这可能是去这里的方式？

如何使用 Dask 优化搜索参数空间？（没有交叉验证）

这是代码（这里没有DASK）：

这是我的使用方法：

问题是：

1. 我在这里正确使用 Dask 吗？
2. 我是否正确地将数据取回给客户端？有没有更有效的方法来做到这一点？
3. 有没有办法在工人身上寻找最佳配对？

PS最近我发布了相关问题，但有不同的问题（使用Dask分布式自定义参数搜索类期间的thread.lock）。我已经解决了它，并将很快发布答案并将关闭该问题。

我有以下带有以下列的熊猫数据框

所有列都包含整数。我不想对它们执行一些操作并使用 dask 数据框运行它们。这就是我所做的。

但我收到以下错误

难道我做错了什么？在纯熊猫中它工作得很好，但是对于许多行来说它会变慢（尽管它们适合内存）所以我想并行化计算。

我已将输入数据存储为 S3 上的单个大文件。我希望 Dask 自动切分文件，分发给工作人员并管理数据流。因此使用分布式集合的想法，例如袋子。

在每个工作人员上，我都有一个从文件中读取数据的命令行工具 (Java)。因此，我想将一整块数据写入文件，调用外部 CLI/代码来处理数据，然后从输出文件中读取结果。这看起来像是处理批量数据而不是一次记录。

解决这个问题的最佳方法是什么？是否可以将分区写入工作人员的磁盘并将其作为一个整体进行处理？

PS。保留在分布式集合模型中也没有必要，但也是可取的，因为对数据的其他操作可能是更简单的 Python 函数，它们逐条处理数据。

我注意到使用只有进程和没有保姆的 dask 工作人员与使用保姆相比，性能下降了 5/6 倍。这是预期的行为吗？

由于工人的状态，我想在没有保姆的情况下运行 dask。我很欣赏在工人中拥有状态是不可取的，但它超出了我的控制范围（第 3 方图书馆）。

或者，如果我与保姆一起运行 dask 工作人员，我可以捕获工作人员故障/重新启动并重新初始化工作人员吗？

由于特定于我的构建系统的原因，我目前无法运行 dask-scheduler CLI - 我必须从代​​码启动调度程序（示例在 dask.distributed 文档中提供）。还有一种方法可以为调度程序启动和自定义（即选择端口）Web GUI 吗？

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

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

我正在尝试使用 Dask 和 Airflow 实现数据管道。我希望能够向现有 DAG 添加/删除节点，类似于 NodeJS 中的中间件。我的想法是腌制数据框，以便下一步可以在腌制之前对其进行任何转换并传递到下一步。但是使用 Dask 并行处理，下一个节点可以分配给任何工作人员。我正在考虑在本地拥有调度程序和一些工作人员。当我有一份大工作时，我可以解雇一些 EC2 工作人员来处理这项工作。有什么建议吗？

我正在使用 Apache 和 Django 开发一个 Web 应用程序应用程序，用户在其中与数据模型（包装到 Python 中的 C++ 实现）进行交互。

为了避免在每次用户操作后在文件或数据库中加载/保存数据，只要用户连接到应用程序，我更喜欢将数据模型保留在内存中。到目前为止，数据模型都存储在附加到 Web 服务的变量中。由于在 Apache 下运行的 Python 有时会出现奇怪的行为，我更喜欢将用户操作执行到单独的 python 进程中，今天在同一台服务器上，也许明天在不同的节点上。

我的印象是分布式计算库（dispy、dask 分布式）无法将内存连接到节点。有没有人对我可以使用哪些库有解决方案/想法？