问题标签 [dask-delayed]

对象是否dask.delayed由集群上的 dask 分发？

另外，它的任务图的执行是否也分布在集群上？

pandas.DataFrame.copy API可以在dask.DataFrame中完全模仿，使用下面的代码吗？

是简单拷贝还是深拷贝？我怎样才能做其他类型的副本？

或者我是否必须执行以下操作？

第二个代码片段会完全解决我的问题，还是有一些警告？

如果我有一个 dask 数据框 df. 现在我对其进行一些计算。

数学上，

df1 = f1(df)

df2 = f2(df1)

df3 = f3(df1)

现在，如果我跑步df2.compute()，现在之后，如果我跑步df1.compute()。如何阻止 dask 重新计算结果df1？

以另一种情况为例，如果我运行df3.compute()，则df2.compute()。我如何告诉 dask 在运行时使用df1（在 中计算df3.compute()）的已计算值df2.compute()？

我正在尝试用我自己的分区替换 Series dask 分区。我在这篇文章中使用了@MRocklin给出的代码片段。

除了 dask_df 在我的情况下是一个系列之外，我已经完成了完全相同的操作。我收到以下错误：

我有 1000 个正则表达式模式，我必须在 9000 个字符串中的每一个中进行搜索。使用 pandas 列表的正常蛮力方法需要 25 分钟来完成相同的任务。我使用了 dask 的延迟函数来并行化整个函数。完成任务花了9分钟。我需要更快的速度。如何利用 dask 数组或 dask 数据框来完成任务？或者有什么更快的方法吗？

我有一个超过 35000 行的熊猫系列。我想使用 dask 使其更高效。但是，我的 dask 代码和 pandas 代码都花费了相同的时间。最初“ser”是熊猫系列，而fun1和fun2是在系列的各个行中执行模式匹配的基本函数。

熊猫：

黎明：

在检查 cpu 核心的状态时，我发现并不是所有的核心都被使用了。只有一个核心习惯了 100%。

有什么方法可以使用 dask 使系列代码更快，或者在串联执行 Dask 操作时利用 cpu 的所有内核？

我期望从使用 Dask 延迟的简单示例中了解到，我基本上可以通过如下几个函数调用从 scikit-learn 复制 gridsearchcv。看起来模型永远不适合（model.fit（...）），因为循环的其余部分继续（pred（...））？

我如何嵌套函数有问题吗？我知道 dask 有 gridsearchcv，但问题是我的真实模型是多输入 Keras LSTM，您不能将 3d 数组作为“X”传递。该代码在没有 Dask 的情况下串行工作正常。

这是一个可重现的小例子：

添加#1

这是第二次尝试，错误仍然存​​在。

添加#2

事实证明，Keras / TF 有一个问题会导致 Dask 之外的错误。我将在一个单独的问题中解决这个问题。因此，我将 Keras 模型换成了 Xgboost 模型，以便为此目的正确设置 Dask。

这是那个代码。我确实发现我需要注释掉在 mean_squared_error 位中延迟的对 Dask 的调用。

我一直在尝试编写自定义类，用于Preprocessing跟随Feature selection和Machine Learning算法。

我(preprocessing only)使用@delayed. 但是当我从教程中读到相同的内容时，可以使用Client. 它引起了两个问题。

作为脚本运行。不像 Jupyter 笔记本

第一个问题：

我在 Jupyter Notebook 中尝试了同样的方法，没有在不同的终端中运行任何调度程序或工作程序。有效！！

现在，我用 1 个调度程序和 2 个工作人员触发了 3 个终端，并将其更改为Client('IP')在脚本中。错误已解决，此行为的任何原因。

第二个问题：

问题标题中提到的错误。将client = Client('IP')作为参数传递给构造函数并将使用self.client.submit的东西传递给集群。但失败并显示错误消息

错误：没有模块名称“diya_info”

这是代码：

主文件

迪亚信息.py

这是完整的堆栈跟踪：

如果我取消注释@delayed和更多评论，它会起作用。client但是如何通过传入作为参数来使其工作。想法是对我要编写的所有库使用相同的客户端。

更新 1： 我second problem通过删除@staticmethod装饰器并将函数放置在fit closure. 但是有什么问题@staticmethod，这些装饰器是为非自我相关的东西设计的，对吧？

这是diya_info.py：

有没有办法做到这一点@staticmethod。我对解决这个问题的方式感觉不好。仍然没有任何线索Problem 1

我正在使用 dask.distributed 运行模拟。我的模型是在延迟函数中定义的，我堆叠了几个实现。此代码片段中给出了我所做工作的简化版本：

如果我想运行大量模拟，我会收到以下警告：

据我了解（从this和this question），警告提出的方法有助于将大数据输入函数。但是，我的输入都是标量值，因此它们不应占用近 3MB 的内存。即使该函数run_model()根本不接受任何参数（因此没有传递参数），我也会收到相同的警告。

我还查看了任务图，看看是否有某个步骤需要加载大量数据。对于三个实现，它看起来像这样：

因此，在我看来，每个实现都是单独处理的，这应该使要处理的数据量保持在低水平。

我想了解产生大对象的实际步骤是什么，以及我需要做什么才能将其分解成更小的部分。

我是 DASK 的新手，想测试在集群上运行 DASK。集群有一个头服务器和几个其他节点。登录头服务器后，我可以通过简单的 ssh 无需密码即可进入其他节点。我想运行一个简单的函数来迭代一个大数组。该函数定义如下。就是将 dt64 转换为 numpy datetime 对象。

import xarray as xr import numpy as np from dask import compute, delayed import dask.multiprocessing from datetime import datetime, timedelta def converdt64(dt64): ts = (dt64 - np.datetime64('1970-01-01T00:00:00Z')) / np.timedelta64(1, 's') return datetime.utcfromtimestamp(ts)

然后在终端上，我通过应用这个函数来迭代一个大小为 N 的 1D 数组。

values = [delayed(convertdt64)(x) for x in arraydata] results1 = compute(*values,scheduler='processes’)

这在头服务器上使用了一些内核，并且它可以工作，尽管速度很慢。然后我尝试使用客户端在集群的几个节点上启动该功能，如下所示：

from dask.distributed import Client client = Client("10.140.251.254:8786 ») results = compute(*values, scheduler='distributed’)

它根本不起作用。有一些警告和一条错误消息，如下所示。

我也尝试了 dask.bag 并且收到了相同的错误消息。集群上的并行计算不起作用的原因可能是什么？是由于某些服务器/网络配置，还是我对 DASK 客户端的使用不正确？在此先感谢您的帮助 ！

最好的祝愿

香农 X