问题标签 [dask]

TL;DR：我如何将分布式读取中的元数据（解析期间的错误）收集到 dask 数据帧集合中。

我目前有一个专有文件格式，用于输入 dask.DataFrame。我有一个接受文件路径并返回 pandas.DataFrame 的函数，dask.DataFrame 在内部使用它成功地将多个文件加载到同一个 dask.DataFrame。

直到最近，我还在使用自己的代码将几个 pandas.DataFrames 合并为一个，现在我正在改用 dask。解析文件格式时，我可能会遇到错误和某些条件，我想记录并与 dask.DataFrame 对象关联为元数据（日志、数据来源等）。

重要的是要注意，在合理的情况下，我大量使用 MultiImdices（13 个索引级别，3 个列级别）。对于描述整个数据框而不是特定行的元数据，我使用的是属性。

使用自定义函数，我可以将元数据与实际 DataFrame 一起传递到元组中。使用 pandas，我可以将其添加到 _metadata 字段并作为 DataFrame 对象的属性。使用 dask 框架时，如何从单独的 pandas.DataFrame 对象收集元数据？

谢谢！

我正在使用 dask 自定义图表

我如何打印图形定义

例如考虑这个grpah

我想通过调用类似的东西打印出上述描述

我知道 dot_graph(dask,'image.png') 但我想获得文字描述

我有一些相当大的 csv 文件（~10gb）并且想利用 dask 进行分析。但是，根据我设置要读取的 dask 对象的分区数量，我的 groupby 结果会发生变化。我的理解是 dask 利用分区来获得核外处理的好处，但它仍然会返回适当的 groupby 输出。情况似乎并非如此，我正在努力找出需要哪些替代设置。下面是一个小例子：

groupby 步骤是否仅在每个分区内运行，而不是查看完整的数据帧？在这种情况下，设置 npartitions=1 是微不足道的，而且它似乎不会对性能产生太大影响，但是由于 read_csv 会自动设置一定数量的分区，您如何设置调用以确保 groupby 结果准确？

谢谢！

I have a directory containing n h5 file each of which has m image stacks to filter. For each image, I will run the filtering (gaussian and laplacian) using dask parallel arrays in order to speed up the processing (Ref to Dask). I will use the dask arrays through the apply_parallel() function in scikit-image.
I will run the processing on a small server with 20 cpus.

I would like to get an advice to which parallel strategy will make more sense to use:

1) Sequential processing of the h5 files and all the cpus for dask processing
2) Parallel processing of the h5 files with x cores and use the remaining 20-x to dask processing.
3) Distribute the resources and parallel processing the h5 files, the images in each h5 files and the remaining resources for dask.

thanks for the help!

我正在研究一个基本上可以归结为求解矩阵方程的项目

其中A是一个矩阵，其尺寸大约为 10 000 000 到 2000 （我想最终在两个方向上增加它）。

A显然不适合内存，因此必须并行化。我通过解决A.T.dot(A).dot(x) = A.T.dot(d)来做到这一点。A.T到 2000 年的维度将是 2000。它可以通过将A和d分成块来计算，A_i并且d_i沿着行计算A_i.T.dot(A_i)和A_i.T.dot(d_i)，然后将它们相加。非常适合并行化。我已经能够使用多处理模块来实现这一点，但是 1）A由于内存使用，很难进一步扩展（在两个维度上都增加），以及 2）不漂亮（因此不容易维护）。

Dask 似乎是一个非常有前途的库，可以解决这两个问题，我也做了一些尝试。我的A矩阵计算起来很复杂：它基于大约 15 个不同的数组（大小等于 中的行数A），其中一些用于迭代算法来评估相关的勒让德函数。当chunk很小（10000行）时，构建任务图需要很长时间，并且占用大量内存（内存增加与迭代算法的调用不谋而合）。当 chunk 较大（50000 行）时，计算前的内存消耗会小很多，但在计算时会迅速耗尽A.T.dot(A)。我已经尝试过cache.Chest，但它显着减慢了计算速度。

任务图一定很大很复杂——调用A._visualize()崩溃。使用更简单A的矩阵，它可以直接执行此操作（请参阅@MRocklin 的回复）。有没有办法让我简化它？

任何有关如何解决此问题的建议将不胜感激。

作为一个玩具示例，我尝试了

这也失败了，耗尽了所有内存，只有一个内核处于活动状态。使用chunks = (2e3, 1e5)时，所有内核几乎立即启动，但MemoryError会在 1 秒内出现（我当前的计算机上有 15 GB）。chunks = (2e3, 1e4)更有希望，但它最终也消耗了所有内存。

编辑：我删除了玩具示例测试，因为尺寸错误，并更正了其余的尺寸。正如@MRocklin 所说，它确实适用于正确的尺寸。我添加了一个我现在认为与我的问题更相关的问题。

Edit2：这是我试图做的一个非常简化的例子。我相信，问题在于定义A.

这似乎导致了一个非常复杂的任务图，甚至在计算开始之前就占用了大量内存。

我现在已经解决了这个问题，方法是将递归放在一个带有numpy数组的函数中，或多或少地做A = x.map_blocks(...).

作为第二点，一旦我有了A矩阵任务图，A.T.dot(A)直接计算似乎确实会产生一些内存问题（内存使用情况不是很稳定）。因此，我明确地分块计算它，并对结果求和。即使有这些变通方法，dask 在速度和可读性方面也有很大的不同。

我有一个非常大的 netCDF 文件，我正在使用 python 中的 netCDF4 读取它

我不能一次全部读取这个文件，因为它的尺寸（1200 x 720 x 1440）对于整个文件来说太大而无法一次在内存中。第 1 个维度代表时间，接下来的 2 个维度分别代表纬度和经度。

然而，一次读一年是极其缓慢的。对于以下用例，我如何加快速度？

- 编辑

块大小为 1

1. 我可以读取一系列年份：nc_file.variables[variable_name][0:100, :, :]

2. 有几个用例：

   以年为单位：

   /li>

我有一个 json 文件目录，我试图将其转换为 dask DataFrame 并将其保存到 castra。它们之间有 200 个包含 O(10**7) json 记录的文件。代码非常简单，主要遵循教程示例。

我在 32 核机器上运行它，但代码仅 100% 使用一个核。我对文档的理解是这段代码是并行执行的。为什么不是？我是不是误会了什么？

我们正在考虑使用 dask，尤其是它的惰性计算和 dag 功能。

我们有一个中等复杂的计算 dag，输入未知。我们希望能够提前构建它，然后在不同的输入上使用它。

我认为我们可以使用 dict / tuple 接口来做到这一点：

然后我们就有dsk了便携的 dag，可以x用我们想要的任何东西替换。（实际上，我们最初不需要将其包含在上面）。

但是我们可以这样做dask.imperative吗？我的初步结果表明我们无法达到x：

有办法吗？

Dask FAQ 提到了奇异值分解 (SVD)，http ://dask.pydata.org/en/latest/faq.html 。但我在 API 中没有看到 SVD。我已经安装了 PyPi 的最新版本 0.8.0。

谢谢，卡尔

在 Dask 中，生成器何时转换为列表，或者它们通常是懒惰地消耗？例如，使用代码：

我得到以下输出，其中生成器已（合理地）转换为列表供我检查：

虽然合理，但它不同于我通常期望生成器在 Python 中的行为方式，后者需要显式转换为列表。

那么，在计算fold( ) 时，是生成器result.compute()的输入参数 ，还是已经将其转换为列表？xadd_to_count

我对列表很长的情况很感兴趣，因此惰性评估更有效，比如 b1 = db.from_sequence([10**6]*10).

我猜我也可以用 解决上述问题bag.frequencies，但我对惰性评估和有效减少也有类似的担忧。

Dask 是否有一个基本方面我不是在摸索，或者我只是懒惰，我在哪里可以查看代码来自己解决这个问题？