问题标签 [zarr]

我正在寻找使用阅读远程 zarr 商店xarray.open_mfdataset()

我得到一个zarr.errors.GroupNotFoundError: group not found at path ''. 追溯在底部。

我可以使用打开它xr.open_zarr

如果我要在本地下载 zarr 商店，它可以正常工作

open_mfdataset远程 zarr 商店的引用

当我运行时，zarr.open('result.zarr', mode='r')我收到以下错误：

FSPathExistNotDir: 路径存在但不是目录: %r

根据位于https://zarr.readthedocs.io/en/stable/tutorial.html#persistent-arrays的 Zarr 文档中的示例，这个 zarr.open() 函数应该返回一个 zarr.core.Array：

在我的情况下，该功能如何zarr.open()寻找目录？

我想阅读https://hrrrzarr.s3.amazonaws.com/index.html#sfc/20210208/20210208_00z_anl.zarr/的远程 zarr 商店。zarr 商店的信息位于https://mesowest.utah.edu/html/hrrr/zarr_documentation/zarrFileVariables.html

我能够读取一个变量，但它似乎没有捕获与变量关联的坐标或属性（我很可能缺少 kwargs to open_mfdatasetor open_zarr）。因为有不同级别的嵌套，我不确定通过的正确路径是什么

将数据集写入xarrayAWS S3 会花费惊人的大量时间，即使实际上没有使用compute=False.

这是一个例子：

即使没有实际计算，与 S3 存储桶位于同一区域的 EC2 也to_zarr需要大约 6 秒的时间。

查看调试日志，似乎正在进行相当多的重定向，因为默认区域 inaiobotocore设置为us-east-2while bucket is in eu-central-1。

如果我首先手动将默认区域放入环境变量中

然后所需时间下降到3.5 秒左右。

所以我的问题是：

1. 有没有办法将该区域传递给fsspec（或s3fs）？我已经尝试添加s3_additional_kwargs={"region":"eu-central-1"}到该get_mapper方法，但没有做任何事情。

2. xarray有没有比上面（with ）更好的方式与 S3 上的 zarr 交互fsspec？

版本：

TLDR：

如何有效地使用dask-distributed将多个支持的数据集写入 AWS S3 上的dask存储？xarrayzarr

详情：

我有一个工作流，它采用 S3 上的栅格数据集列表并生成一个 dask-array 支持的 xarray 数据集。

我需要遍历多个组，对于每个组，工作流都会获取属于该组的栅格数据集并生成相应的 xarray 数据集。

现在我想将数据集中的数据写入 S3 上的 zarr 存储（同一个存储，仅使用group参数）。

顺序处理的伪代码如下所示：

这很好用，一旦to_zarr执行，数据就会被加载并存储在 S3 上，任务并行运行。

现在我想使用dask.distribuited. 这是我尝试过的以及遇到的问题：

1.用于.to_zarr(..., compute=False)收集延迟任务列表

这原则上有效，但速度很慢。创建一个任务大约需要 3-4 秒，我需要运行 100 多次，在实际开始任何计算之前需要 4-5 分钟。

2.把它包起来dask.delayed

这极大地加快了任务的创建速度，但是写入 zarr 存储并没有在工作人员之间分配，而是处理任务的工作人员在加载任务完成后收集所有数据并将其写入 zarr。

3.包装to_zarr自定义函数并将其传递给client.submit

这看起来是最有希望的选择。我刚刚将to_zarr调用包装在一个自定义函数中，可以从工作人员调用：

这样做worker_client会将编写任务放回调度程序并解决我在上面遇到的问题dask.delayed。

但是，当我按照以下方式重复提交此功能时（我需要这样做 100 多次）

我很快用要处理的任务使调度程序不堪重负。

我能想到的唯一明显的解决方案是分批拆分数据集，只有在前一个完成后才开始一个新的。但是没有更优雅的解决方案吗？或者dask（分布式）中是否有内置功能？

背景

我正在使用 dask 管理数以万计、有时甚至是数十万个作业，每个作业都涉及读取 zarr 数据、以某种方式转换数据以及写出输出（每个作业一个输出）。我正在使用带有 dask 网关的pangeo/daskhub风格的 jupyterhub kubernetes 集群。

数据不能对齐到一个巨大的 dask 支持的 xarray DataArray，相反，我们常见的使用模式是简单地使用 dask 分布式Client.map将函数映射到任务。每个任务都可以在内存中运行。

问题在于，对于包括读取 zarr 数组在内的某些操作，xarray 会自动使用集群调度程序调度每个 I/O 操作，即使该操作是由远程工作人员调用的。这使调度程序必须管理的任务数量成倍增加，有时是一个很大的倍数。如果 zarr 数组有很多块，有时 dask 调度程序会尝试分配读取，从而导致集群内出现大量网络流量，当许多任务尝试排队以读取其块时，这可能会使进度停止同时。

完全有可能这里的正确答案是“不要这样做”，例如，对于如此庞大而复杂的工作，请使用 Argo 或 Kubeflow 之类的东西。但我想看看是否有人对如何使这个工作与 dask 一起工作有想法。

问题

我的问题本质上是在 dask.distributed 任务中运行时是否可以阻止 xarray（或具有本机 dask 支持的另一个库）使用集群的调度程序。

我认为我所看到的理想可能是这样的：

我不确定这是否会涉及更改 xarray、zarr 或 dd.get_client() 或其他东西的行为。

MRE

可以调整上述内容以获得可测试的示例。目标是看不到除主映射功能之外的任何任务。我在 jupyterlab ipython notebook 中运行了以下命令，并使用 dask-labextension 观看了任务（调度程序仪表板显示相同的结果）

进口

测试文件设置

函数定义

创建客户端并查看仪表板

映射作业

清理（​​如果再次运行）

全面清理

请注意，尽管只有两个作业，但仍安排了数千个任务。

我正在使用具有以下（在许多其他软件包中）的 conda 环境：

在最后一句话的“ region”论点的描述中指出： 。我无法在参考文档或其他任何地方找到此参考 - 我开始怀疑它不存在。谁能给我指出这个所谓文件的位置？xarray.Dataset.to_zarrSee “Appending to existing Zarr stores” in the reference documentation for full details

我正在尝试使用多线程来加快读写 Zarr 文件的速度。例如，如果我可以将数组存储在 5 个块中，有没有办法使用每个块的线程来加快从磁盘读取和写入数组的速度（可能使用 ThreadSynchronizer() 和同步器参数？）。我只想加快读/写速度。我不想并行计算。我知道这可以用 dask 完成。谢谢。

我正在尝试N从存储在 Zarr 组中的所有子组中读取行，并且使用 for 循环非常慢，所以有没有办法使用 Dask（我对 Dask 完全陌生，找不到任何类似于我想要的东西做）来实现这种相对快速且内存效率高的方式

一些代码片段可以帮助我尝试做的事情。

因此，我没有使用 for 循环，而是想知道是否有一种方法可以轻松提取N行，然后我可以简单地将其用作下游工作的 2D 矩阵。

或者：通过连接所有这些子组而不消耗太多内存，知道如何在磁盘上创建二维数组（可能作为 Zarr 存储）也很好。

我正在打开一个 zarr 文件，然后将其重新分块，然后将其写回另一个 zarr 商店。然而，当我打开它备份它不尊重我之前写的块大小。这是 jupyter 的代码和输出。知道我在这里做错了什么吗？

它正在恢复到原始分块，就好像我没有完全覆盖它或更改一些需要更改的其他设置一样。