问题标签 [dask-delayed]

我一直在试图弄清楚如何在 dask 数组的切片上执行函数。例如，如果我创建以下 dask 数组：

并定义函数：

执行

按预期工作，但如果我尝试将这些结果分配回 x，则函数失败：

抛出 NotImplementedError: Item assignment with not supported 有没有一种方法可以解决这个问题来执行此操作？谢谢你，

如何使用 dask 并行运行具有不同列维度的 SQL 查询？以下是我的尝试：

但是，这会导致引发以下错误：

DatabaseError: sql 执行失败: "SQL QUERY" ORA-01013: 用户请求取消当前操作无法回滚

然后不久之后出现另一个错误：

MultipleInstanceError：正在创建多个不兼容的 TerminalInteractiveShell 子类实例。

sql_to_run 是不同 sql 查询的列表

有什么建议或指点吗？？谢谢！

更新 9.7.18

认为这更像是我没有足够仔细地阅读文档的情况。事实上，在 loadsql 函数之外的 con 是导致问题的原因。以下是现在似乎按预期工作的代码更改。

这是我正在做的事情的摘要：

起初，我通过普通的多处理和熊猫包来做到这一点：

步骤 1. 获取我要阅读的文件名列表

步骤 2. 遍历列表

代码工作正常，但速度很慢。

完成该任务大约需要 1000 秒。

library(parallel)与使用和parSapply功能的 R 程序进行比较。

R 程序只需要大约 160 秒。

然后我尝试使用以下代码使用 dask.delayed 和 dask.dataframe：

步骤 1. 获取我要阅读的文件名列表

步骤 2. 遍历列表

这一次，我发现如果我在 dask 代码和 pandas 代码中同时注释掉第 2.3 步，dask 的运行速度会比普通的 pandas 和多处理快得多。

但是如果我调用 to_csv 方法，那么 dask 和 pandas 一样慢。

有什么解决办法吗？

谢谢

我有两个数据集（称为 satdata 和 atmosdata）。Atmosdata 在纬度和经度上均匀网格化。Atmosdata的维度为（纬度：713，级别：37，经度：1440，时间：72），总大小为12GB。Atmosdata 有几个变量，例如温度、湿度等。湿度具有（时间、水平、纬度、经度）的形状。

Satdata包含卫星观测数据，维度为(across_track: 90, channel: 3, time: 32195)，数据点数为90*3*32195=8692650。Across_track 表示卫星 FOV 越过轨道位置。卫星数据在纬度/经度上不是均匀网格化的。例如，satdata.latitude 的维度为（时间、通道、cross_track），与 satdata.longitude、satdata.sft 相同。

Atmosdata 和 satdata 中的“时间”变量包含同一天的时间，但在这两个数据集中具有不同的值。我需要找到与 satdata 具有相同纬度、经度和时间的 atmosdata（例如湿度和温度）。

为了实现这一点，我遍历 satdata 以找到每次观察的位置和时间；然后我找到相应的 atmosdata（首先是离卫星数据位置最近的网格，然后插值到卫星时间）。最后，我将所有迭代生成的 atmosdata 连接到一个数据集中。

通过使用一小段数据，我的代码的一部分如下。

匹配函数如下：

1）当我启动代码时，它可以工作。但是，如果我不在代码中使用小数据大小，而是使用我的原始数据（具有上述尺寸），然后我启动计算并出现错误。

2）如果我使用其他维度的数据集，satdata（across_track：90，channel：3，时间：100）和atmosdata（纬度：71，水平：37，经度：1440，时间：72），计算时间很长时间。我想我的编码并不是使用 DASK 快速解决这个问题的最佳选择。

2）有没有比使用 for 循环更好的方法？为了快速计算，for 循环可能不会利用 DASK 吗？

3）将satdata分块，然后在satdata块中找到纬度和经度的限制，然后根据此限制对atmosdata进行分块，最后将匹配函数应用于每个satdata和atmosdata块？如果这是个好主意，我还不知道如何手动迭代每个 satdata 块......

4) 该函数使用两个参数，satdata 和 atmosdata。由于这两个数据集可能相当大（atmosdata 为 12G），那么计算会更慢吗？

5）在我必须在选择中使用 .value 的函数中，当使用大输入数据时，这是否会使计算变慢？

提前致谢 ！

此致

小妮

我使用了一个自定义字典之类的对象来轻松存储 Dask 图的结果，但是使用生成的对象来计算 Dask 图，不计算它的孩子。

是否可以更改自定义对象，使其 Dask 能够遍历并计算其子对象？

一个例子：

结果输出：

编辑：基于 MRocklin 回答的解决方案

结果：

（它仍然显示defaultdict，因为__repr__没有被正确覆盖）

我正在尝试从一组压缩的 CSV 文件创建一个 dask 数据框。阅读问题，似乎 dask 需要使用 dask.distributed delayed()

zip_dict = {'log20160201': zipfile.ZipFile filename='/my_directory/zip_files/log20160201.zip' mode='r', 'log20160218': zipfile.ZipFile filename='/my_directory/zip_files/log20160218.zip' 的示例内容模式='r'}

d_rows 的样本内容 = [Delayed('read_csv-c05dc861-79c3-4e22-8da6-927f5b7da123'), Delayed('read_csv-4fe1c901-44b4-478b-9c11-4a80f7a639e2')]

返回的错误是： ValueError: Invalid file path or buffer object type: class 'list'

我正在尝试将迭代器通过（非标准）类似文件的对象传递给dask.delayed函数。当我尝试时compute()，我从 dask 收到以下消息，以及下面的回溯。

源代码的对应部分如下所示：

vf是类文件对象，并vf.fetch(..)返回文件中存在的记录的迭代器（这是一个VCF 文件，我正在使用pysam库来读取它）。我希望这提供了足够的背景。

来自的消息dask显示迭代发生在函数调用期间而不是函数内部，这让我相信传递迭代器可能是不合适的。所以我做了一个快速检查sum(range(..))，这似乎有效。现在我很困惑，我错过了什么？

为此提供一个最小的工作示例有点困难。但也许以下内容会有所帮助。

1. 从这里下载一个 VCF 文件（和它的索引）：说，ALL.chrY*vcf.gz{,.tbi}
2. pip3 install --user pysam
3. 打开文件：vf = VariantFile('/path/to/file.vcf.gz', mode='r')
4. 像这样的迭代器：vf.fetch("Y", 2_600_000, 2_700_000)
5. 对于延迟函数，您可以有一个空循环。

我有一个向我返回数据框的函数。我正在尝试通过使用 dask 来并行使用此功能。
我将数据帧的延迟对象附加到列表中。但是，我的代码的运行时间在有和没有 dask.delayed 的情况下是相同的。
我使用 functools 中的 reduce 函数pd.merge来合并我的数据框。
关于如何提高运行时的任何建议？
可视化图表和代码如下。

操作的可视化图表

我正在尝试在 Python 中使用 Dask 合并许多大型数据集，以避免加载问题。我想保存为.csv合并文件。事实证明，这项任务比想象的要难：

我用两个数据集组合了一个玩具示例然后我使用的代码如下：

如果我使用dd_all.to_csv('*.csv')，我只需打印出两个原始数据集。如果我使用dd_all.to_csv('name.csv')我会收到一条错误消息，指出该文件不存在。( FileNotFoundError: [Errno 2] No such file or directory: 'C:\\Users\\Me\\Working directory\\name.csv\\1.part')

我可以检查是否dd_all.compute()已成功创建使用合并数据集。

我正在尝试查看 Dask 是否适合我的项目，并编写了一些非常简单的测试用例来研究它的性能。但是，Dask 需要相对较长的时间来简单地执行延迟初始化。

我已经研究了Dask For Loop In Parallel和parallel dask for loop 比常规循环慢？我尝试将大小增加到 100 万个项目，但是虽然常规循环大约需要一秒钟，但 dask 循环永远不会结束。在等待半小时完成 add_vald 的延迟初始化后，我将其杀死。

我不确定这里出了什么问题，非常感谢您提供的任何见解。谢谢！