Generate annual/yearly zarr stores from hourly ERA5 data on NCAR’s GDEX - ARCO ERA-5 Interactive Visualization Cookbook

Overview and Warning: Please Read¶

ERA5 data on NCAR’s Geoscience Data Exchange (GDEX) is stored in hourly NetCDF files. But, kerchunk reference files have been created for pressure level data and zarr stores have been created for the surface variables to convert this dataset into an ARCO collection.
In this notebook, we will read hourly data from NCAR’s publicly accessible ERA5 collection using an intake-ESM catalog, compute the annual means and store the result using zarr stores.
If you don’t have write permision to save to NCAR’s GDEX, please save the result to your local folder.
If you need annual means for the following variables, please don’t run the commented cells in this notebook. The data has already been calculated and can be accessed via https from https://data.gdex.ucar.edu/special_projects/pythia_2024/annual_means/
1. Air temperature at 2 m/ VAR_2T
Otherwise, please run this script once to generate the annual means.

Prerequisites¶

Concepts	Importance	Notes
Intro to Xarray	Necessary
Intro to Intake	Necessary
Understanding of Zarr	Helpful

Time to learn: 30 minutes

Imports¶

import glob
import re
import matplotlib as plt
import numpy as np
import scipy as sp
import xarray as xr
import intake
import intake_esm
import pandas as pd

import dask
from dask.distributed import Client, performance_report,LocalCluster
from dask_jobqueue import PBSCluster

######## File paths ################
gdex_scratch   = "/glade/derecho/scratch/harshah"
era5_catalog = 'https://data.gdex.ucar.edu/d633000/catalogs/d633000-https.json'
annual_means      = '/gdex/data/special_projects/pythia_2024/annual_means/' #Posix path, only used for writing data 
annual_means_https= 'https://data.gdex.ucar.edu/special_projects/pythia_2024/annual_means/'

print(era5_catalog)

https://data.gdex.ucar.edu/d633000/catalogs/d633000-https.json

Create a Dask cluster¶

Dask Introduction¶

Dask is a solution that enables the scaling of Python libraries. It mimics popular scientific libraries such as numpy, pandas, and xarray that enables an easier path to parallel processing without having to refactor code.

There are 3 components to parallel processing with Dask: the client, the scheduler, and the workers.

The Client is best envisioned as the application that sends information to the Dask cluster. In Python applications this is handled when the client is defined with client = Client(CLUSTER_TYPE). A Dask cluster comprises of a single scheduler that manages the execution of tasks on workers. The CLUSTER_TYPE can be defined in a number of different ways.

There is LocalCluster, a cluster running on the same hardware as the application and sharing the available resources, directly in Python with dask.distributed.
In certain JupyterHubs Dask Gateway may be available and a dedicated dask cluster with its own resources can be created dynamically with dask.gateway.
On HPC systems dask_jobqueue is used to connect to the HPC Slurm and PBS job schedulers to provision resources.

The dask.distributed client python module can also be used to connect to existing clusters. A Dask Scheduler and Workers can be deployed in containers, or on Kubernetes, without using a Python function to create a dask cluster. The dask.distributed Client is configured to connect to the scheduler either by container name, or by the Kubernetes service name.

Select the Dask cluster type¶

The default will be LocalCluster as that can run on any system.

If running on a HPC computer with a PBS Scheduler, set to True. Otherwise, set to False.

USE_PBS_SCHEDULER = False

If running on Jupyter server with Dask Gateway configured, set to True. Otherwise, set to False.

USE_DASK_GATEWAY = False

Python function for a PBS cluster

# Create a PBS cluster object
def get_pbs_cluster():
    """ Create cluster through dask_jobqueue.   
    """
    from dask_jobqueue import PBSCluster
    cluster = PBSCluster(
        job_name = 'dask-pythia-24',
        cores = 1,
        memory = '4GiB',
        processes = 1,
        local_directory = gdex_scratch + '/dask/spill',
        resource_spec = 'select=1:ncpus=1:mem=8GB',
        queue = 'casper',
        walltime = '1:00:00',
        #interface = 'ib0'
        interface = 'ext'
    )
    return cluster

Python function for a Gateway Cluster

def get_gateway_cluster():
    """ Create cluster through dask_gateway
    """
    from dask_gateway import Gateway

    gateway = Gateway()
    cluster = gateway.new_cluster()
    cluster.adapt(minimum=2, maximum=4)
    return cluster

Python function for a Local Cluster

def get_local_cluster():
    """ Create cluster using the Jupyter server's resources
    """
    from distributed import LocalCluster, performance_report
    cluster = LocalCluster()    
    cluster.scale(6)
    return cluster

Python logic to select the Dask Cluster type

This uses True/False boolean logic based on the variables set in the previous cells

# Obtain dask cluster in one of three ways

if USE_PBS_SCHEDULER:
    cluster = get_pbs_cluster()
elif USE_DASK_GATEWAY:
    cluster = get_gateway_cluster()
else:
    cluster = get_local_cluster()

# Connect to cluster
from distributed import Client
client = Client(cluster)

# Display cluster dashboard URL
cluster

Find data using intake catalog¶

era5_cat = intake.open_esm_datastore(era5_catalog)
era5_cat

---------------------------------------------------------------------------
SSLCertVerificationError                  Traceback (most recent call last)
File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/aiohttp/connector.py:1337, in TCPConnector._wrap_create_connection(self, addr_infos, req, timeout, client_error, *args, **kwargs)
   1336             kwargs["ssl_shutdown_timeout"] = self._ssl_shutdown_timeout
-> 1337         return await self._loop.create_connection(*args, **kwargs, sock=sock)
   1338 except cert_errors as exc:

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/asyncio/base_events.py:1199, in BaseEventLoop.create_connection(self, protocol_factory, host, port, ssl, family, proto, flags, sock, local_addr, server_hostname, ssl_handshake_timeout, ssl_shutdown_timeout, happy_eyeballs_delay, interleave, all_errors)
   1196         raise ValueError(
   1197             f'A Stream Socket was expected, got {sock!r}')
-> 1199 transport, protocol = await self._create_connection_transport(
   1200     sock, protocol_factory, ssl, server_hostname,
   1201     ssl_handshake_timeout=ssl_handshake_timeout,
   1202     ssl_shutdown_timeout=ssl_shutdown_timeout)
   1203 if self._debug:
   1204     # Get the socket from the transport because SSL transport closes
   1205     # the old socket and creates a new SSL socket

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/asyncio/base_events.py:1237, in BaseEventLoop._create_connection_transport(self, sock, protocol_factory, ssl, server_hostname, server_side, ssl_handshake_timeout, ssl_shutdown_timeout)
   1236 try:
-> 1237     await waiter
   1238 except:

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/asyncio/sslproto.py:581, in SSLProtocol._on_handshake_complete(self, handshake_exc)
    580 else:
--> 581     raise handshake_exc
    583 peercert = sslobj.getpeercert()

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/asyncio/sslproto.py:563, in SSLProtocol._do_handshake(self)
    562 try:
--> 563     self._sslobj.do_handshake()
    564 except SSLAgainErrors:

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/ssl.py:969, in SSLObject.do_handshake(self)
    968 """Start the SSL/TLS handshake."""
--> 969 self._sslobj.do_handshake()

SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: self-signed certificate in certificate chain (_ssl.c:1082)

The above exception was the direct cause of the following exception:

ClientConnectorCertificateError           Traceback (most recent call last)
File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/fsspec/implementations/http.py:447, in HTTPFileSystem._info(self, url, **kwargs)
    445 try:
    446     info.update(
--> 447         await _file_info(
    448             self.encode_url(url),
    449             size_policy=policy,
    450             session=session,
    451             **self.kwargs,
    452             **kwargs,
    453         )
    454     )
    455     if info.get("size") is not None:

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/fsspec/implementations/http.py:861, in _file_info(url, session, size_policy, **kwargs)
    860 elif size_policy == "get":
--> 861     r = await session.get(url, allow_redirects=ar, **kwargs)
    862 else:

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/aiohttp/client.py:900, in ClientSession._request(self, method, str_or_url, params, data, json, cookies, headers, skip_auto_headers, auth, allow_redirects, max_redirects, compress, chunked, expect100, raise_for_status, read_until_eof, proxy, proxy_auth, timeout, verify_ssl, fingerprint, ssl_context, ssl, server_hostname, proxy_headers, trace_request_ctx, read_bufsize, auto_decompress, max_line_size, max_field_size, max_headers, middlewares, upload_tracker)
    899 try:
--> 900     resp = await handler(req)
    901 # Client connector errors should not be retried

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/aiohttp/client.py:854, in ClientSession._request.<locals>._connect_and_send_request(req)
    853 try:
--> 854     conn = await self._connector.connect(
    855         req, traces=traces, timeout=real_timeout
    856     )
    857 except asyncio.TimeoutError as exc:

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/aiohttp/connector.py:665, in BaseConnector.connect(self, req, traces, timeout)
    664         await trace.send_connection_create_start()
--> 665 proto = await self._create_connection(req, traces, timeout)
    666 if traces:

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/aiohttp/connector.py:1263, in TCPConnector._create_connection(self, req, traces, timeout)
   1262 else:
-> 1263     _, proto = await self._create_direct_connection(req, traces, timeout)
   1265 return proto

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/aiohttp/connector.py:1641, in TCPConnector._create_direct_connection(self, req, traces, timeout, client_error)
   1640 assert last_exc is not None
-> 1641 raise last_exc

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/aiohttp/connector.py:1610, in TCPConnector._create_direct_connection(self, req, traces, timeout, client_error)
   1609 try:
-> 1610     transp, proto = await self._wrap_create_connection(
   1611         self._factory,
   1612         timeout=timeout,
   1613         ssl=sslcontext,
   1614         addr_infos=addr_infos,
   1615         server_hostname=server_hostname,
   1616         req=req,
   1617         client_error=client_error,
   1618     )
   1619 except (ClientConnectorError, asyncio.TimeoutError) as exc:

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/aiohttp/connector.py:1339, in TCPConnector._wrap_create_connection(self, addr_infos, req, timeout, client_error, *args, **kwargs)
   1338 except cert_errors as exc:
-> 1339     raise ClientConnectorCertificateError(req.connection_key, exc) from exc
   1340 except ssl_errors as exc:

ClientConnectorCertificateError: Cannot connect to host data.gdex.ucar.edu:443 ssl:True [SSLCertVerificationError: (1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: self-signed certificate in certificate chain (_ssl.c:1082)')]

The above exception was the direct cause of the following exception:

FileNotFoundError                         Traceback (most recent call last)
Cell In[10], line 1
----> 1 era5_cat = intake.open_esm_datastore(era5_catalog)
      2 era5_cat

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/intake_esm/core.py:113, in esm_datastore.__init__(self, obj, progressbar, sep, registry, read_csv_kwargs, columns_with_iterables, storage_options, **intake_kwargs)
    111     self.esmcat = ESMCatalogModel.from_dict(obj)
    112 else:
--> 113     self.esmcat = ESMCatalogModel.load(
    114         obj, storage_options=self.storage_options, read_csv_kwargs=read_csv_kwargs
    115     )
    117 self.derivedcat = registry or default_registry
    118 self._entries = {}

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/intake_esm/cat.py:239, in ESMCatalogModel.load(cls, json_file, storage_options, read_csv_kwargs)
    236 json_file = str(json_file)  # We accept Path, but fsspec doesn't.
    237 _mapper = fsspec.get_mapper(json_file, **storage_options)
--> 239 with fsspec.open(json_file, **storage_options) as fobj:
    240     data = json.loads(fobj.read())
    241     if 'last_updated' not in data:

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/fsspec/core.py:105, in OpenFile.__enter__(self)
    102 mode = self.mode.replace("t", "").replace("b", "") + "b"
    104 try:
--> 105     f = self.fs.open(self.path, mode=mode)
    106 except FileNotFoundError as e:
    107     if has_magic(self.path):

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/fsspec/spec.py:1417, in AbstractFileSystem.open(self, path, mode, block_size, cache_options, compression, **kwargs)
   1415 else:
   1416     ac = kwargs.pop("autocommit", not self._intrans)
-> 1417     f = self._open(
   1418         path,
   1419         mode=mode,
   1420         block_size=block_size,
   1421         autocommit=ac,
   1422         cache_options=cache_options,
   1423         **kwargs,
   1424     )
   1425     if not ac and "r" not in mode:
   1426         self.transaction.files.append(f)

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/fsspec/implementations/http.py:386, in HTTPFileSystem._open(self, path, mode, block_size, autocommit, cache_type, cache_options, size, **kwargs)
    384 kw.update(kwargs)
    385 info = {}
--> 386 size = size or info.update(self.info(path, **kwargs)) or info["size"]
    387 session = sync(self.loop, self.set_session)
    388 if block_size and size and info.get("partial", True):

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/fsspec/asyn.py:217, in sync_wrapper.<locals>.wrapper(*args, **kwargs)
    214 @functools.wraps(func)
    215 def wrapper(*args, **kwargs):
    216     self = obj or args[0]
--> 217     return sync(self.loop, func, *args, **kwargs)

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/fsspec/asyn.py:106, in sync(loop, func, timeout, *args, **kwargs)
    104     raise FSTimeoutError from return_result
    105 elif isinstance(return_result, BaseException):
--> 106     raise return_result
    107 else:
    108     return return_result

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/fsspec/asyn.py:59, in _runner(event, coro, result, timeout)
     57     coro = asyncio.wait_for(coro, timeout=timeout)
     58 try:
---> 59     result[0] = await coro
     60 except Exception as ex:
     61     result[0] = ex

File ~/micromamba/envs/ERA5_interactive/lib/python3.14/site-packages/fsspec/implementations/http.py:460, in HTTPFileSystem._info(self, url, **kwargs)
    457     except Exception as exc:
    458         if policy == "get":
    459             # If get failed, then raise a FileNotFoundError
--> 460             raise FileNotFoundError(url) from exc
    461         logger.debug("", exc_info=exc)
    463 return {"name": url, "size": None, **info, "type": "file"}

FileNotFoundError: https://data.gdex.ucar.edu/d633000/catalogs/d633000-https.json

# Filter the catalog for variables with 'temp' in their long name
cat_filtered = era5_cat.search(long_name='temp*')
cat_filtered.df.head()

Search for variable of interest¶

############ Access temperature data ###########
temp_cat = era5_cat.search(variable='VAR_2T')
temp_cat.df

# Define the xarray_open_kwargs with a compatible engine
xarray_open_kwargs = {'engine': 'zarr'}

%%time
dset_temp = temp_cat.to_dataset_dict(xarray_open_kwargs=xarray_open_kwargs)

dset_temp.keys()

temp_2m = dset_temp['VAR_2T.2t'].VAR_2T
temp_2m

Resample hourly data to annual data and save¶

# temp_2m_annual = temp_2m.resample(time='YE').mean()
# temp_2m_annual

# temp_2m_annual = temp_2m_annual.chunk({'latitude':721,'longitude':1440})
# temp_2m_annual = temp_2m_annual.drop_isel({'time':-1}) # Drop 2024 data
# temp_2m_annual

Save annual mean to annual_means folder within gdex_data
You may not have permissions to write to this folder.
If so, please replace the folder path in the cell below with your own path where you have write permissions

# %%time
# temp_2m_annual.to_dataset().to_zarr(annual_means + 'temp_2m_annual_1940_2023.zarr',mode='w')

Open the saved file and make a test plot¶

temp_2m_annual = xr.open_zarr(annual_means_https + 'temp_2m_annual_1940_2023.zarr').VAR_2T
temp_2m_annual

%%time
temp_2m_annual.isel(time=0).plot()

Close up the cluster¶

cluster.close()