Streaming Data - intake-esgf Documentation

In addition to the transfer of entire files, data may be streamed to the user as it is required by their script. The benefit is that if only a small portion of the data is to be used, we avoid downloading the whole file. At the time of this writing, ESGF indices only contain OPeNDAP access information. However, as we consider expanding support, the below interface will extend to other streaming/cloud-ready technologies such as Zarr stores, kerchunk, and VirtualiZarr.

To demonstrate this functionality, consider the following search for some future surface air temperature data from the UKESM model.

cat = ESGFCatalog().search(
    experiment_id="ssp585",
    source_id="UKESM1-0-LL",
    variable_id="tas",
    frequency="mon",
)
cat.remove_ensembles()

Summary information for 1 results:
mip_era                 [CMIP6]
activity_drs      [ScenarioMIP]
institution_id           [MOHC]
source_id         [UKESM1-0-LL]
experiment_id          [ssp585]
member_id            [r1i1p1f2]
table_id                 [Amon]
variable_id               [tas]
grid_label                 [gn]
dtype: object

To harvest the OPeNDAP access link from the index nodes, you tell the package that you prefer_streaming=True. Not all files will have this capability, but if they do, then this will tell intake-esgf to use them. Also, in this example we do not need any cell measures and so we will disable that in this search.

dsd = cat.to_dataset_dict(prefer_streaming=True, add_measures=False)

---------------------------------------------------------------------------TimeoutError                              Traceback (most recent call last)File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/urllib3/connection.py:204, in HTTPConnection._new_conn(self)
    203 try:
--> 204     sock = connection.create_connection(
    205         (self._dns_host, self.port),
    206         self.timeout,
    207         source_address=self.source_address,
    208         socket_options=self.socket_options,
    209     )
    210 except socket.gaierror as e:
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/urllib3/util/connection.py:85, in create_connection(address, timeout, source_address, socket_options)
     84 try:
---> 85     raise err
     86 finally:
     87     # Break explicitly a reference cycle
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/urllib3/util/connection.py:73, in create_connection(address, timeout, source_address, socket_options)
     72     sock.bind(source_address)
---> 73 sock.connect(sa)
     74 # Break explicitly a reference cycle
TimeoutError: timed out
The above exception was the direct cause of the following exception:
ConnectTimeoutError                       Traceback (most recent call last)File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/urllib3/connectionpool.py:788, in HTTPConnectionPool.urlopen(self, method, url, body, headers, retries, redirect, assert_same_host, timeout, pool_timeout, release_conn, chunked, body_pos, preload_content, decode_content, **response_kw)
    787 # Make the request on the HTTPConnection object
--> 788 response = self._make_request(
    789     conn,
    790     method,
    791     url,
    792     timeout=timeout_obj,
    793     body=body,
    794     headers=headers,
    795     chunked=chunked,
    796     retries=retries,
    797     response_conn=response_conn,
    798     preload_content=preload_content,
    799     decode_content=decode_content,
    800     **response_kw,
    801 )
    803 # Everything went great!
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/urllib3/connectionpool.py:493, in HTTPConnectionPool._make_request(self, conn, method, url, body, headers, retries, timeout, chunked, response_conn, preload_content, decode_content, enforce_content_length)
    492 try:
--> 493     conn.request(
    494         method,
    495         url,
    496         body=body,
    497         headers=headers,
    498         chunked=chunked,
    499         preload_content=preload_content,
    500         decode_content=decode_content,
    501         enforce_content_length=enforce_content_length,
    502     )
    504 # We are swallowing BrokenPipeError (errno.EPIPE) since the server is
    505 # legitimately able to close the connection after sending a valid response.
    506 # With this behaviour, the received response is still readable.
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/urllib3/connection.py:500, in HTTPConnection.request(self, method, url, body, headers, chunked, preload_content, decode_content, enforce_content_length)
    499     self.putheader(header, value)
--> 500 self.endheaders()
    502 # If we're given a body we start sending that in chunks.
File ~/.asdf/installs/python/3.14.0/lib/python3.14/http/client.py:1333, in HTTPConnection.endheaders(self, message_body, encode_chunked)
   1332     raise CannotSendHeader()
-> 1333 self._send_output(message_body, encode_chunked=encode_chunked)
File ~/.asdf/installs/python/3.14.0/lib/python3.14/http/client.py:1093, in HTTPConnection._send_output(self, message_body, encode_chunked)
   1092 del self._buffer[:]
-> 1093 self.send(msg)
   1095 if message_body is not None:
   1096 
   1097     # create a consistent interface to message_body
File ~/.asdf/installs/python/3.14.0/lib/python3.14/http/client.py:1037, in HTTPConnection.send(self, data)
   1036 if self.auto_open:
-> 1037     self.connect()
   1038 else:
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/urllib3/connection.py:331, in HTTPConnection.connect(self)
    330 def connect(self) -> None:
--> 331     self.sock = self._new_conn()
    332     if self._tunnel_host:
    333         # If we're tunneling it means we're connected to our proxy.
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/urllib3/connection.py:213, in HTTPConnection._new_conn(self)
    212 except SocketTimeout as e:
--> 213     raise ConnectTimeoutError(
    214         self,
    215         f"Connection to {self.host} timed out. (connect timeout={self.timeout})",
    216     ) from e
    218 except OSError as e:
ConnectTimeoutError: (<HTTPConnection(host='esgf-data04.diasjp.net', port=80) at 0x7e5bacc2e120>, 'Connection to esgf-data04.diasjp.net timed out. (connect timeout=10)')
The above exception was the direct cause of the following exception:
MaxRetryError                             Traceback (most recent call last)File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/requests/adapters.py:696, in HTTPAdapter.send(self, request, stream, timeout, verify, cert, proxies)
    695 try:
--> 696     resp = conn.urlopen(
    697         method=request.method,
    698         url=url,
    699         body=request.body,  # type: ignore[arg-type]  # urllib3 stubs don't accept Iterable[bytes | str]
    700         headers=request.headers,  # type: ignore[arg-type]  # urllib3#3072
    701         redirect=False,
    702         assert_same_host=False,
    703         preload_content=False,
    704         decode_content=False,
    705         retries=self.max_retries,
    706         timeout=resolved_timeout,
    707         chunked=chunked,
    708     )
    710 except (ProtocolError, OSError) as err:
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/urllib3/connectionpool.py:842, in HTTPConnectionPool.urlopen(self, method, url, body, headers, retries, redirect, assert_same_host, timeout, pool_timeout, release_conn, chunked, body_pos, preload_content, decode_content, **response_kw)
    840     new_e = ProtocolError("Connection aborted.", new_e)
--> 842 retries = retries.increment(
    843     method, url, error=new_e, _pool=self, _stacktrace=sys.exc_info()[2]
    844 )
    845 retries.sleep()
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/urllib3/util/retry.py:543, in Retry.increment(self, method, url, response, error, _pool, _stacktrace)
    542     reason = error or ResponseError(cause)
--> 543     raise MaxRetryError(_pool, url, reason) from reason  # type: ignore[arg-type]
    545 log.debug("Incremented Retry for (url='%s'): %r", url, new_retry)
MaxRetryError: HTTPConnectionPool(host='esgf-data04.diasjp.net', port=80): Max retries exceeded with url: /thredds/dodsC/esg_dataroot/CMIP6/ScenarioMIP/MOHC/UKESM1-0-LL/ssp585/r1i1p1f2/Amon/tas/gn/v20190507/tas_Amon_UKESM1-0-LL_ssp585_r1i1p1f2_gn_201501-204912.nc (Caused by ConnectTimeoutError(<HTTPConnection(host='esgf-data04.diasjp.net', port=80) at 0x7e5bacc2e120>, 'Connection to esgf-data04.diasjp.net timed out. (connect timeout=10)'))
During handling of the above exception, another exception occurred:
ConnectTimeout                            Traceback (most recent call last)Cell In[4], line 1
----> 1 dsd = cat.to_dataset_dict(prefer_streaming=True, add_measures=False)
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/intake_esgf/catalog.py:728, in ESGFCatalog.to_dataset_dict(self, prefer_streaming, globus_endpoint, globus_path, add_measures, minimal_keys, ignore_facets, separator, quiet, open_kwargs)
    725     open_kwargs = intake_esgf.conf["default_open_kwargs"]
    727 # get the paths, passing along options
--> 728 paths = self.to_path_dict(
    729     prefer_streaming=prefer_streaming,
    730     globus_endpoint=globus_endpoint,
    731     globus_path=globus_path,
    732     minimal_keys=False,
    733     ignore_facets=ignore_facets,
    734     separator=separator,
    735     quiet=quiet,
    736 )
    738 # load paths into xarray objects (also log files accessed)
    739 exceptions = []
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/intake_esgf/catalog.py:612, in ESGFCatalog.to_path_dict(self, prefer_streaming, globus_endpoint, globus_path, minimal_keys, ignore_facets, separator, quiet)
    610 # get and partition file info based on access method
    611 raw_infos = self._get_file_info(separator=separator, quiet=quiet)
--> 612 infos, dsd = base.partition_infos(
    613     raw_infos,
    614     prefer_streaming,
    615     prefer_globus,
    616 )
    618 # optionally use globus to transfer what we can
    619 tasks = []
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/intake_esgf/base.py:197, in partition_infos(infos, prefer_streaming, prefer_globus)
    193 if links:
    194     # Try to find a streaming link from the fastest servers we have
    195     # seen while downloading. If this fails, we revert to https.
    196     try:
--> 197         link = select_streaming_link(links, df_rate)
    198         if key not in ds:
    199             ds[key] = []
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/intake_esgf/base.py:112, in select_streaming_link(links, df_rate)
    108 # Return the first of these links whose html page returns a valid response.
    109 # This is particular to OPENDAP and will need rethought for other virtual
    110 # methods.
    111 for link in links:
--> 112     resp = requests.head(link, timeout=10)
    113     try:
    114         resp.raise_for_status()
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/requests/api.py:114, in head(url, **kwargs)
    103 r"""Sends a HEAD request.
    104 
    105 :param url: URL for the new :class:`Request` object.
   (...)    110 :rtype: requests.Response
    111 """
    113 kwargs.setdefault("allow_redirects", False)
--> 114 return request("head", url, **kwargs)
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/requests/api.py:71, in request(method, url, **kwargs)
     67 # By using the 'with' statement we are sure the session is closed, thus we
     68 # avoid leaving sockets open which can trigger a ResourceWarning in some
     69 # cases, and look like a memory leak in others.
     70 with sessions.Session() as session:
---> 71     return session.request(method=method, url=url, **kwargs)
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/requests/sessions.py:651, in Session.request(self, method, url, params, data, headers, cookies, files, auth, timeout, allow_redirects, proxies, hooks, stream, verify, cert, json)
    646 send_kwargs = {
    647     "timeout": timeout,
    648     "allow_redirects": allow_redirects,
    649 }
    650 send_kwargs.update(settings)
--> 651 resp = self.send(prep, **send_kwargs)
    653 return resp
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/requests/sessions.py:784, in Session.send(self, request, **kwargs)
    781 start = preferred_clock()
    783 # Send the request
--> 784 r = adapter.send(request, **kwargs)
    786 # Total elapsed time of the request (approximately)
    787 elapsed = preferred_clock() - start
File ~/checkouts/readthedocs.org/user_builds/intake-esgf/checkouts/latest/.venv/lib/python3.14/site-packages/requests/adapters.py:717, in HTTPAdapter.send(self, request, stream, timeout, verify, cert, proxies)
    714 if isinstance(e.reason, ConnectTimeoutError):
    715     # TODO: Remove this in 3.0.0: see #2811
    716     if not isinstance(e.reason, NewConnectionError):
--> 717         raise ConnectTimeout(e, request=request)
    719 if isinstance(e.reason, ResponseError):
    720     raise RetryError(e, request=request)
ConnectTimeout: HTTPConnectionPool(host='esgf-data04.diasjp.net', port=80): Max retries exceeded with url: /thredds/dodsC/esg_dataroot/CMIP6/ScenarioMIP/MOHC/UKESM1-0-LL/ssp585/r1i1p1f2/Amon/tas/gn/v20190507/tas_Amon_UKESM1-0-LL_ssp585_r1i1p1f2_gn_201501-204912.nc (Caused by ConnectTimeoutError(<HTTPConnection(host='esgf-data04.diasjp.net', port=80) at 0x7e5bacc2e120>, 'Connection to esgf-data04.diasjp.net timed out. (connect timeout=10)'))

At this point, the dataset dictionary is returned but you will notice that no file download messages were received. The OPeNDAP access link was passed to the xarray constructor. We now proceed with our analysis as if the data is local. In this example, we wish to see what future temperatures will be under the SSP585 scenario over my hometown.

ds = dsd["tas"]
ds = ds.sel(lat=35.96, lon=-83.92 + 360, method="nearest")

Now we can plot this trace using matlotlib. When the xarray dataset needs data, it uses the OPeNDAP protocol to stream just the time trace at the specific location.

fig, ax = plt.subplots(figsize=(10, 3))
ds["tas"].plot(ax=ax);

This can be a very fast alternative if the data volume is relatively low. If you want to verify that data has indeed been streamed and not accessed locally, you may print the session log and look at what was accessed.

print(cat.session_log())

If you look towards the bottom of that log, you will see that a https link was accessed in place of a local file. Note that if a local file is present in your local cache, we will use that file even if you have preferred to use streaming.