在 Amazon S3 上使用 Zarr 将地理空间查询延迟从几分钟减少到几秒

地理空间数据，包括许多气候和天气数据集，通常由政府和非营利组织以压缩文件格式发布，例如 网络通用数据表 (NetCDF) 或 Gridded Binary (GRIB)。 这些格式中有许多是在云前世界中设计的，用户可以下载整个文件在本地计算机上进行分析。随着地理空间数据集复杂性和规模的持续增长，将文件放在一个地方，虚拟查询数据并仅下载本地所需的子集，这样既省时又更具成本效益。

与传统文件格式不同，云原生 Zarr 格式旨在虚拟高效地访问保存在中心位置的压缩数据块，例如来自亚马逊网络服务 (亚马逊云科技) 的 亚马逊简单存储服务 (Amazon S3)。

在本演练中，学习如何使用亚马逊 SageMa ker 笔记本 和 亚马逊云科技 Fargate 集群将 NetCDF 数据集转换为 Zar r，并查询生成的 Zarr 存储，从而将时间序列查询所需的时间从几分钟缩短到几秒。

传统的地理空间数据格式

地理空间栅格数据文件将数据点与经纬度网格中的像元相关联。此类数据至少是二维的，但如果多个数据变量与每个像元相关联（例如，温度和风速），或者随着时间的推移记录变量或与海拔等附加维度相关联，则会增长到 N 维。

多年来，已经设计了许多文件格式来存储此类数据。1985 年创建的 GRIB 格式将数据存储为压缩二维记录的集合，每条记录对应一个时间步。但是，GRIB 文件不包含支持直接访问记录的元数据；检索单个变量的时间序列需要顺序访问每条记录。

NetCDF 于 1990 年推出，在 GRIB 的一些缺陷基础上进行了改进。NetCDF 文件包含支持更高效的数据索引和检索的元数据，包括远程打开文件和虚拟浏览其结构，而无需将所有内容加载到内存中。NetCDF 数据可以作为区块存储在文件中，从而实现对不同区块的并行读取和写入。自 2002 年起，NetCDF4 使用 分层数据格式版本 5 (HDF5 ) 作为后端，提高了区块上并行 I/O 的速度。但是，由于单个文件中包含多个区块，NetCDF4 数据集的查询性能仍可能受到限制。

使用 Zarr 生成云原生地理空间数据

Zarr 文件格式将 N 维数据组织为组（数据集）和数组（数据集中的变量）的层次结构。每个数组的数据都存储为一组压缩的二进制块，每个区块都写入自己的文件。

将 Zarr 与 Amazon S3 搭配使用可实现毫秒级延迟的区块访问。区块读取和写入可以并行运行并根据需要进行扩展，既可以通过来自单台计算机的多线程或进程，也可以通过分布式计算资源（例如在 亚马逊弹性容器服务 (Amazon ECS) 中运行的容 器）进行扩展。将 Zarr 数据存储在 Amazon S3 中并应用分布式计算不仅可以避免数据下载时间和成本，而且通常是分析地理空间数据的唯一选择，因为这些数据太大而无法放入单台计算机的 RAM 或磁盘上。

最后，Zarr 格式包含可访问的元数据，允许用户远程打开和浏览数据集，并仅在本地下载所需的数据子集，例如分析的输出。下图显示了包含一个数组的 Zarr 组中包含的元数据文件和区块的示例。

图 1。包含一个 Zarr 数组的 Zarr 组。

Zarr 与开源 P angeo 技术堆栈 中的其他 Python 库紧密集成， 用于分析科学数据，例如 用于协调分布式并行计算作业的 Dask 和 用作处理标记 N 维 数组数据的接口的 xarray 。

Zarr 性能优势

亚马逊云科技 上的开放数据 注册表 包含许多广泛使用的地理空间数据集的 NetCDF 和 Zarr 版本，包括 ECMWF 再分析 v5 (ERA5) 数据集。 ERA5 包含自 2008 年以来在 31 千米纬度网格上多个高度的每小时气压、风速和气温数据。 我们比较了从ERA5的netCDF和Zarr版本中查询这三个变量在给定纬度和经度上一年、五、九和十三年的历史数据的平均时间。

图 2。比较使用具有 10 个工作线程（8 个 vCPU、16 个 GM 内存）的 Dask 集群查询三个 ERA5 变量的平均时间。

请注意，这些测试并不是文件格式的真正一对一比较，因为每种格式使用的区块大小不同，结果也可能因网络条件而异。但是，它们与基准研究一致， 基准研究 表明 ，与NetCDF相比，Zarr具有显著的性能优势。

尽管 亚马逊云科技 上的开放数据注册表包含许多流行的地理空间数据集的 Zarr 版本，但这些数据集的 Zarr 分块策略可能未针对组织的数据访问模式进行优化，或者组织可能拥有大量采用 NetCDF 格式的内部数据集。在这些情况下，他们可能希望从 NetCDF 文件转换为 Zarr。

搬到扎尔

将 NetCDF 文件转换为 Zarr 有几种选择：P angeo-forge 配方、使用 Python re chunker 库和使用 xarray。

Pangeo-Forge 项目提供了预制 配方 ，可以让你将一些 GRIB 或 NetCDF 数据集转换为 Zarr，而无需深入研究特定格式的底层细节。

为了更好地控制转换过程，你可以使用 rechunker 和 xarray 库。Rechunker 可以高效地操作保存在永久存储中的分块数组。由于 rechunker 使用中间文件存储器并且专为并行执行而设计，因此它可用于将大于工作内存的数据集从一种文件格式转换为另一种文件格式（例如从 NetCDF 转换为 Zarr），同时更改底层区块的大小。这使其成为将大型 NetCDF 数据集转换为优化的 Zarr 格式的理想之选。Xarray 可用于将较小的数据集转换为 Zarr 或将数据从 NetCDF 文件附加到现有 Zarr 存储，例如时间序列中的其他数据点（rechunker 目前不支持）。

最后，虽然从技术上讲不是 从 NetCDF 转换到 Zarr 的选项，但 K erchunk python 库创建了索引文件，允许像访问 Zarr 存储一样访问 NetCDF 数据集，与单独使用 NetCDF 相比，无需创建新的 Zarr 数据集即可显著提高性能。

这篇博文的其余部分提供了使用 rechunker 和 xarray 将 NetCDF 数据集转换为 Zarr 的分步说明。

解决方案概述

该解决方案部署在 连接 互联网网关 的 亚马逊虚拟私有云 (Amazon VPC) 中。在私有子网内，它既部署了在 亚马逊云科技 Far gate 上运行的 Dask 集群，又部署了用于向 Dask 集群提交任务的 亚马逊 SageMaker 笔记本 实例。 在公有子网中， NAT 网关 为来自 Dask 集群或笔记本实例的连接提供互联网接入（例如，在安装库时）。最后， 网络负载均衡器 (NLB) 允许访问 Dask 仪表板来监控 Dask 作业的进度。VPC 中的资源通过亚马逊 S3 VP C 终端节点访问 亚马逊 S3 。

图 3。解决方案的架构图。

解决方案演练

以下是执行本演练所需步骤的简要概述，我们将对其进行更详细的介绍：

1. 克隆 GitHub 存储库 。
2. 使用 亚马逊云科技 云开发套件 (亚马逊云科技 CDK) 部署基础设施。
3. （可选）启用 Dask 控制面板。
4. 在 SageMaker 笔记本实例上设置 Jupyter 笔记本。
5. 使用 Jupyter 笔记本将 NetCDF 文件转换为 Zarr。

前四个步骤大部分是自动完成的，大约需要 45 分钟。设置完成后，运行 Jupyter 笔记本中的单元应该需要 30 分钟。

先决条件

在开始本演练之前，您应该具备以下先决条件：

• 一个 亚马逊云科技 账户
• 亚马逊云科技 身份和访问管理 (IAM) 有权执行本指南中的所有步骤
• Python3 和 亚马逊云科技 CDK 安装 在您要部署解决方案的计算机上。

部署解决方案

第 1 步：克隆 GitHub 存储库

在终端中，运行以下代码来克隆 包含解决方案的 GitHub 存储库 。

git clone https://github.com/aws-samples/convert-netcdf-to-zarr.git

cd convert-netcdf-to-zarr

python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

cdk deploy

cd SageMaker

git clone https://github.com/aws-samples/convert-netcdf-to-zarr.git

cd convert-netcdf-to-zarr/notebooks

conda env create --name zarr_py310_nb -f environment.yml

DASK_SCHEDULER_URL = "Dask-Scheduler.local-dask:8786"
client = Client(DASK_SCHEDULER_URL)

ds_nc = xr.open_mfdataset(nc_files_map, engine='h5netcdf', chunks={}, parallel=True)

# returns [“time”, “lat”, “lon”]
dims = ds_nc.dims.keys()

new_chunksizes = (1080, 90, 90)
new_chunksizes_dict = dict(zip(dims, new_chunksizes))

ds_zarr = rechunk(
    ds_nc, 
    target_chunks={
      var_name: new_chunksizes_dict, 
      'time': None,'lat': None,'lon': None
    },
    max_mem='15GB',
    target_store = zarr_store, 
    temp_store = zarr_temp
).execute()

cdk destroy