Unexpected Local Data Transfer During xarray Preprocessing on Dask Cluster

[lbesnard]

TL;DR

When running a script on a remote Dask cluster (e.g., via Coiled), memory on the local machine (which starts the cluster) is unexpectedly used. This happens I think after opening NetCDF files from S3 via s3fs with xarray.open_mfdataset and applying a preprocess function — even though all operations are expected to run remotely on the cluster.
This is part of a script which afterwards writes the data to Zarr from the cluster (not shown here). But the local machine can download/upload Gb of data as well as using tens of Gb of memory.

Background

I initially got some help from @phofl who raised this fsspec #1747

I also tried to get some insights here:

• https://discourse.pangeo.io/t/xarray-operations-e-g-preprocess-running-locally-post-open-mfdataset-instead-of-on-dask-distributed-cluster/4637
• Xarray operations (e.g., preprocess) running locally (post open_mfdataset) instead of on Dask distributed cluster dask/distributed#8913 (comment)

Apparently the trick was to use the following options:

s3_fs = s3fs.S3FileSystem(
    anon=True, default_fill_cache=False, default_cache_type="readahead"
)

However, this never fixed it for me. My local machine always ends up being used. I've kind of given up, and used smaller batches of files to process at once, but it defeats the purpose of using a cluster.

@martindurant asked me in this comment for a MVCE, but there were far too many annoying bugs on xarray which seemed to have been fixes now.

The MVCE example below can run on any machine since the files on S3 are on an anonymous bucket. I'm using coiled in it, but a Fargate cluster would show the same behaviour of memory and bandwith consumption. This example use a bit more than 6 Gb of local memory I think. If the preprocess function is disabled, memory is still being used.

Maybe it's a dask/xarray problem, and not an s3fs one, but I'm not sure. The only thing I know is that I can replicate this all the time, with any dataset I'm playing with. Depending on the dataset, my local memory/bandwith might explode. I'm very surprised I'm the only one experiencing it

MVCE

fsspec                        2025.5.1
s3fs                          2025.5.1
xarray                    2025.7.1
dask                      2025.5.1
dask-cloudprovider        2024.9.1
dask-expr                 2.0.0

#!/usr/bin/env python3
from functools import partial

import s3fs
import xarray as xr
from coiled import Cluster
from dask.distributed import Client
import numpy as np
import numcodecs
import os


def create_fileset(s3_paths, s3_fs=None):
    """
    Create a fileset from S3 objects specified by a list of full S3 paths.

    Args:
        s3_paths (str or list[str]): Either a single full S3 path (e.g., 's3://bucket_name/object_key')
                                     or a list of full S3 paths.

    Returns:
        list[file-like object]: List of file-like objects representing each object in the fileset.
    """
    if s3_fs is None:
        s3_fs = s3fs.S3FileSystem(
            anon=True, default_fill_cache=False, default_cache_type="readahead"
        )

    if isinstance(s3_paths, str):
        s3_paths = [s3_paths]

    if not isinstance(s3_paths, list):
        raise ValueError("Invalid input format. Expecting either str or list[str].")

    # Create a fileset by opening each file
    fileset = [s3_fs.open(file) for file in s3_paths]

    return fileset


def preprocess_xarray(ds, dataset_config=None):
    append_dim = "TIME"
    dest_name = "filename"

    filename_placeholder = "UNKNOWN_FILENAME.nc"
    filename = filename_placeholder

    try:
        var = next(var for var in ds)  # get the first variable
        if hasattr(ds[var], "encoding") and "source" in ds[var].encoding:
            filename = os.path.basename(ds[var].encoding["source"])
        elif hasattr(ds, "encoding") and "source" in ds.encoding:
            filename = os.path.basename(ds.encoding["source"])

        print(f"Filename found in ds.encoding: {filename}")
    except Exception as e:
        print(
            f"Original filename not available in the xarray dataset: {e} - will default to use {filename_placeholder}"
        )
        filename = filename_placeholder

    length = ds.sizes[append_dim]
    string_array = np.full(
        length, filename, dtype=object
    )  # having da.full could break things when appending to existing zarr. need to investigate why. default to using numpy array instead

    ds[dest_name] = (append_dim, string_array)
    ds[dest_name].encoding = {
        "_FillValue": "",
        "dtype": object,
        "compressor": None,
        "filters": None,
        "object_codec": numcodecs.VLenUTF8(),  # crucial!
    }

    return ds


class GenericHandler:
    def _open_mfds(
        self, partial_preprocess, drop_vars_list, batch_files, engine="scipy"
    ):
        open_mfdataset_params = {
            "engine": engine,
            "parallel": True,
            "preprocess": partial_preprocess,
            "data_vars": "all",
            "concat_characters": True,
            "mask_and_scale": True,
            "decode_cf": True,
            "decode_times": True,
            "use_cftime": True,
            "decode_coords": True,
            "compat": "override",
            "coords": "minimal",
            "drop_variables": drop_vars_list,
        }

        ds = xr.open_mfdataset(batch_files, **open_mfdataset_params)

        # ds = ds.unify_chunks()
        # ds = ds.sortby("TIME")

        return ds

    def publish_cloud_optimised_fileset_batch(self, s3_file_uri_list):
        self.dataset_config = None
        drop_vars_list = ["wind_speed"]

        partial_preprocess = partial(
            preprocess_xarray, dataset_config=self.dataset_config
        )

        ds = self._open_mfds(
            partial_preprocess,
            drop_vars_list,
            s3_file_uri_list,
        )

       # next part of the code should do a ds.to_zarr() 

prefix = "s3://imos-data/IMOS/SRS/OC/radiometer/VLHJ_Southern-Surveyor/2013/"
s3_file_uri_list += [
    prefix + "IMOS_SRS-OC_F_20130214T000001Z_VLHJ_FV01_DALEC_END-20130214T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130215T000000Z_VLHJ_FV01_DALEC_END-20130215T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130216T000000Z_VLHJ_FV01_DALEC_END-20130216T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130217T000002Z_VLHJ_FV01_DALEC_END-20130217T054432Z.nc",
    prefix + "IMOS_SRS-OC_F_20130218T213111Z_VLHJ_FV01_DALEC_END-20130218T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130219T000001Z_VLHJ_FV01_DALEC_END-20130219T054325Z.nc",
    prefix + "IMOS_SRS-OC_F_20130221T213337Z_VLHJ_FV01_DALEC_END-20130221T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130222T000004Z_VLHJ_FV01_DALEC_END-20130222T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130223T000002Z_VLHJ_FV01_DALEC_END-20130223T235951Z.nc",
    prefix + "IMOS_SRS-OC_F_20130224T000003Z_VLHJ_FV01_DALEC_END-20130224T230710Z.nc",
    prefix + "IMOS_SRS-OC_F_20130226T220957Z_VLHJ_FV01_DALEC_END-20130226T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130227T000001Z_VLHJ_FV01_DALEC_END-20130227T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130228T000000Z_VLHJ_FV01_DALEC_END-20130228T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130301T000002Z_VLHJ_FV01_DALEC_END-20130301T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130302T000001Z_VLHJ_FV01_DALEC_END-20130302T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130303T000000Z_VLHJ_FV01_DALEC_END-20130303T001757Z.nc",
    prefix + "IMOS_SRS-OC_F_20130410T001856Z_VLHJ_FV01_DALEC_END-20130410T061407Z.nc",
    prefix + "IMOS_SRS-OC_F_20130411T002135Z_VLHJ_FV01_DALEC_END-20130411T061533Z.nc",
    prefix + "IMOS_SRS-OC_F_20130412T002318Z_VLHJ_FV01_DALEC_END-20130412T061322Z.nc",
    prefix + "IMOS_SRS-OC_F_20130413T002039Z_VLHJ_FV01_DALEC_END-20130413T061708Z.nc",
    prefix + "IMOS_SRS-OC_F_20130414T001848Z_VLHJ_FV01_DALEC_END-20130414T061153Z.nc",
    prefix + "IMOS_SRS-OC_F_20130415T002210Z_VLHJ_FV01_DALEC_END-20130415T060559Z.nc",
    prefix + "IMOS_SRS-OC_F_20130416T001959Z_VLHJ_FV01_DALEC_END-20130416T055628Z.nc",
    prefix + "IMOS_SRS-OC_F_20130417T001719Z_VLHJ_FV01_DALEC_END-20130417T054750Z.nc",
    prefix + "IMOS_SRS-OC_F_20130418T001858Z_VLHJ_FV01_DALEC_END-20130419T040841Z.nc",
    prefix + "IMOS_SRS-OC_F_20130419T041404Z_VLHJ_FV01_DALEC_END-20130419T051434Z.nc",
    prefix + "IMOS_SRS-OC_F_20130420T002254Z_VLHJ_FV01_DALEC_END-20130420T044248Z.nc",
    prefix + "IMOS_SRS-OC_F_20130421T003058Z_VLHJ_FV01_DALEC_END-20130421T020501Z.nc",
    prefix + "IMOS_SRS-OC_F_20130429T005759Z_VLHJ_FV01_DALEC_END-20130429T031412Z.nc",
    prefix + "IMOS_SRS-OC_F_20130621T022830Z_VLHJ_FV01_DALEC_END-20130621T034358Z.nc",
    prefix + "IMOS_SRS-OC_F_20130622T015425Z_VLHJ_FV01_DALEC_END-20130622T040101Z.nc",
    prefix + "IMOS_SRS-OC_F_20130623T022616Z_VLHJ_FV01_DALEC_END-20130623T035247Z.nc",
    prefix + "IMOS_SRS-OC_F_20130624T023403Z_VLHJ_FV01_DALEC_END-20130624T042341Z.nc",
    prefix + "IMOS_SRS-OC_F_20130625T024217Z_VLHJ_FV01_DALEC_END-20130625T045224Z.nc",
    prefix + "IMOS_SRS-OC_F_20130626T025409Z_VLHJ_FV01_DALEC_END-20130626T050829Z.nc",
    prefix + "IMOS_SRS-OC_F_20130627T032159Z_VLHJ_FV01_DALEC_END-20130627T051355Z.nc",
    prefix + "IMOS_SRS-OC_F_20130728T003922Z_VLHJ_FV01_DALEC_END-20130728T071533Z.nc",
    prefix + "IMOS_SRS-OC_F_20130729T002107Z_VLHJ_FV01_DALEC_END-20130729T070924Z.nc",
    prefix + "IMOS_SRS-OC_F_20130730T000226Z_VLHJ_FV01_DALEC_END-20130730T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130731T000002Z_VLHJ_FV01_DALEC_END-20130731T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130801T000001Z_VLHJ_FV01_DALEC_END-20130801T235929Z.nc",
    prefix + "IMOS_SRS-OC_F_20130805T212533Z_VLHJ_FV01_DALEC_END-20130805T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130806T000002Z_VLHJ_FV01_DALEC_END-20130806T005357Z.nc",
    prefix + "IMOS_SRS-OC_F_20130807T042802Z_VLHJ_FV01_DALEC_END-20130807T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130808T000000Z_VLHJ_FV01_DALEC_END-20130808T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130809T000000Z_VLHJ_FV01_DALEC_END-20130809T063416Z.nc",
    prefix + "IMOS_SRS-OC_F_20130820T040551Z_VLHJ_FV01_DALEC_END-20130820T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130821T000002Z_VLHJ_FV01_DALEC_END-20130821T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130822T000001Z_VLHJ_FV01_DALEC_END-20130822T235955Z.nc",
    prefix + "IMOS_SRS-OC_F_20130823T000000Z_VLHJ_FV01_DALEC_END-20130823T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130824T000001Z_VLHJ_FV01_DALEC_END-20130824T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130825T000002Z_VLHJ_FV01_DALEC_END-20130825T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130826T000004Z_VLHJ_FV01_DALEC_END-20130826T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130827T000003Z_VLHJ_FV01_DALEC_END-20130827T211558Z.nc",
    prefix + "IMOS_SRS-OC_F_20130828T204304Z_VLHJ_FV01_DALEC_END-20130828T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130829T000000Z_VLHJ_FV01_DALEC_END-20130829T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130830T000003Z_VLHJ_FV01_DALEC_END-20130830T235956Z.nc",
    prefix + "IMOS_SRS-OC_F_20130831T000001Z_VLHJ_FV01_DALEC_END-20130831T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130901T000000Z_VLHJ_FV01_DALEC_END-20130901T235954Z.nc",
    prefix + "IMOS_SRS-OC_F_20130902T000000Z_VLHJ_FV01_DALEC_END-20130902T034340Z.nc",
    prefix + "IMOS_SRS-OC_F_20130927T201704Z_VLHJ_FV01_DALEC_END-20130927T235956Z.nc",
    prefix + "IMOS_SRS-OC_F_20130928T000001Z_VLHJ_FV01_DALEC_END-20130928T235955Z.nc",
    prefix + "IMOS_SRS-OC_F_20130930T203535Z_VLHJ_FV01_DALEC_END-20130930T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20131001T000000Z_VLHJ_FV01_DALEC_END-20131001T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20131002T000003Z_VLHJ_FV01_DALEC_END-20131002T034135Z.nc",
]


cluster_options = {
    "n_workers": [1, 120],
    "scheduler_vm_types": "t3.2xlarge",
    "worker_vm_types": "t3.2xlarge",
    "allow_ingress_from": "me",
    "compute_purchase_option": "spot_with_fallback",
    "worker_options": {"nthreads": 8, "memory_limit": "32GB"},
    "name": "thisIsATest",
}

cluster = Cluster(**cluster_options)

client = Client(cluster)

s3_file_handle_list = create_fileset(s3_file_uri_list)
GenericHandler().publish_cloud_optimised_fileset_batch(s3_file_handle_list)

[martindurant]

Thank you for providing a lot of detail - it will take me some time to get through this.

There are a few things going on, so we may need to do more debugging to find out what's using up your memory.

You say your data is "netCDF" - is this netCDF3 or netCDF4?

Firstly: the file serialisation issue in s3fs we believe is fixed. You should be able to explicitly call pickle.dumps in S3File objects to see (and we could make a test around this). By the way, I recommend "first" as the best caching strategy for getting the metadata only (as opposed to arrays such as the coordinates). You can also turn off the cache altogether.

We should also figure out why file objects are being serialised at all. I am surprised the files should be touched in the client in any way, given "minimal" and "override" options. Someone from xarray will need to think about that.

You suspect that the memory is getting used up due to preprocessing, so it would be good to get the memory diff before/after applying the preprocessing function. How big are the string_array objects?

Another thing to test, is the size of the produced dask graph, e.g., number of tasks. The graph(s) are attributed of the array objects within ds.

[lbesnard]

Note that the examples have a lot of files, as it's more obvious to see network bandwith and mem usage.

TL;DR

I suspect an issue somewhere linked to the type of files NetCDF3/4 as the behaviour is different depending on the engine being used (scipy vs h5netcdf). I can't use the netcdf4 engine (because of s3fs I think, not sure anymore)

NetCDF3, scipy engine with preprocess (same as original code posted above)

I added this bit of code after xr.open_mfdataset

        for var_name, da in ds.data_vars.items():
            if (
                hasattr(da.data, "__dask_graph__")
                and da.data.__dask_graph__() is not None
            ):
                graph = da.data.__dask_graph__()
                n_tasks = len(graph)
                print(f"{var_name}: {n_tasks} tasks")
            else:
                print(f"{var_name}: Not a Dask array or no computation graph")

which showed the following results

[2025-07-16 11:14:07,158][INFO    ][coiled] Using existing cluster: 'thisIsATest (id: 1034505)'
[2025-07-16 11:14:07,158][INFO    ][coiled] Attaching to existing cluster (name: thisIsATest, https://cloud.coiled.io/clusters/1034505 )
relaz: 270 tasks
sza: 270 tasks
pitch: 270 tasks
pitchsd: 270 tasks
roll: 270 tasks
rollsd: 270 tasks
gearpos: 270 tasks
Ed: 270 tasks
Lu: 270 tasks
Lsky: 270 tasks
Ed_quality_control: 270 tasks
Lu_quality_control: 270 tasks
Lsky_quality_control: 270 tasks
filename: Not a Dask array or no computation graph

I tried with the following default_cache_type values {"first", "readahead", "none"}. I got the same behaviour with any of these options.

With the code I initially provided on my first comment, once dask has opened all the files and performed the open_dataset, getattr and preprocess_xarray tasks (names visible on the dask dashboard), not much is happening for a few minutes, then my local computer is progressively increasing its memory usage up to 8 Gb of ram with a spike of 15Gb or more. It also downloads a total of 7.7 Gb (at least I'm at the university with an OK bandwith) , and uploads almost nothing (but on other instances, I've seen my laptop uploading a lot of data back) (I'm monitoring with btop).

NetCDF3, scipy engine without preprocess

I then ran the same code again, but I commented out the line "preprocess": partial_preprocess,.
I'm not sure why, but in this instance, coiled scaled up the workers up to 6. I have no idea why it decided to do so when there is no preprocessing, but I'm just sharing it as this could be a hint.

However, same behaviour as above.

NetCDF4, h5netcdf engine, with or without preprocess

In this section, we have another example, which is pretty much the same, but with NetCDF4 gridded files using the h5netcdf engine.

In this instance, the process runs as it should. No data is sent back to the local machine. So it could be an engine issue, or a fsspec one specific to NetCDF3 files ?

I have other scenarios (not included here) where I can't use the h5netcdf or scipy engine only as I'm trying to open a mix of NetCDF3 and NetCDF4 together using xr.concat after opening single files individually with different engines.

#!/usr/bin/env python3
from functools import partial

import s3fs
import xarray as xr
from coiled import Cluster
from dask.distributed import Client
import numpy as np
import numcodecs
import os
import dask.array as da


def create_fileset(s3_paths, s3_fs=None):
    """
    Create a fileset from S3 objects specified by a list of full S3 paths.

    Args:
        s3_paths (str or list[str]): Either a single full S3 path (e.g., 's3://bucket_name/object_key')
                                     or a list of full S3 paths.

    Returns:
        list[file-like object]: List of file-like objects representing each object in the fileset.
    """
    if s3_fs is None:
        s3_fs = s3fs.S3FileSystem(
            anon=True, default_fill_cache=False, default_cache_type="readahead"
        )

    if isinstance(s3_paths, str):
        s3_paths = [s3_paths]

    if not isinstance(s3_paths, list):
        raise ValueError("Invalid input format. Expecting either str or list[str].")

    # Create a fileset by opening each file
    fileset = [s3_fs.open(file) for file in s3_paths]

    return fileset


def preprocess_xarray(ds, dataset_config=None):
    append_dim = "time"
    dest_name = "filename"

    filename_placeholder = "UNKNOWN_FILENAME.nc"
    filename = filename_placeholder

    try:
        var = next(var for var in ds)  # get the first variable
        if hasattr(ds[var], "encoding") and "source" in ds[var].encoding:
            filename = os.path.basename(ds[var].encoding["source"])
        elif hasattr(ds, "encoding") and "source" in ds.encoding:
            filename = os.path.basename(ds.encoding["source"])

        print(f"Filename found in ds.encoding: {filename}")
    except Exception as e:
        print(
            f"Original filename not available in the xarray dataset: {e} - will default to use {filename_placeholder}"
        )
        filename = filename_placeholder

    length = ds.sizes[append_dim]
    string_array = np.full(
        length, filename, dtype=object
    )  # having da.full could break things when appending to existing zarr. need to investigate why. default to using numpy array instead

    ds[dest_name] = (append_dim, string_array)
    ds[dest_name].encoding = {
        "_FillValue": "",
        "dtype": object,
        "compressor": None,
        "filters": None,
        "object_codec": numcodecs.VLenUTF8(),  # crucial!
    }

    return ds


class GenericHandler:
    def _open_mfds(
        self, partial_preprocess, drop_vars_list, batch_files, engine="h5netcdf"
    ):
        open_mfdataset_params = {
            "engine": engine,
            "parallel": True,
            "preprocess": partial_preprocess,
            "data_vars": "all",
            "concat_characters": True,
            "mask_and_scale": True,
            "decode_cf": True,
            "decode_times": True,
            "use_cftime": True,
            "decode_coords": True,
            "compat": "override",
            "coords": "minimal",
            "drop_variables": drop_vars_list,
        }

        ds = xr.open_mfdataset(batch_files, **open_mfdataset_params)

        # ds = ds.unify_chunks()
        # ds = ds.sortby("TIME")

        return ds

    def publish_cloud_optimised_fileset_batch(self, s3_file_uri_list):
        self.dataset_config = None
        drop_vars_list = ["wind_speed"]

        partial_preprocess = partial(
            preprocess_xarray, dataset_config=self.dataset_config
        )

        ds = self._open_mfds(
            partial_preprocess,
            drop_vars_list,
            s3_file_uri_list,
        )

        for var_name, da in ds.data_vars.items():
            if (
                hasattr(da.data, "__dask_graph__")
                and da.data.__dask_graph__() is not None
            ):
                graph = da.data.__dask_graph__()
                n_tasks = len(graph)
                print(f"{var_name}: {n_tasks} tasks")
            else:
                print(f"{var_name}: Not a Dask array or no computation graph")


s3_file_uri_list = [
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211101111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211102111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211103111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211104111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211105111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211106111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211107111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211108111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211109111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211110111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211111111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211112111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211113111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211114111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211115111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211116111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211117111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211118111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211119111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211120111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211121111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211122111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211123111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211124111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211125111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211126111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211127111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211128111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211129111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211130111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211201111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211202111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211203111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211204111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211205111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211206111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211207111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211208111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211209111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211210111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211211111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211212111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211213111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211214111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211215111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211216111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211217111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211218111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211219111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211220111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211221111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211222111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211223111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211224111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211225111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211226111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211227111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211228111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211229111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211230111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
    "s3://imos-data/IMOS/SRS/SST/ghrsst/L3SM-1dS/dn/2021/20211231111000-ABOM-L3S_GHRSST-SSTfnd-MultiSensor-1d_dn_Southern.nc",
]


cluster_options = {
    "n_workers": [1, 120],
    "scheduler_vm_types": "m7i.xlarge",
    "worker_vm_types": "m7i.2xlarge",
    "allow_ingress_from": "me",
    "compute_purchase_option": "spot_with_fallback",
    "worker_options": {"nthreads": 8, "memory_limit": "32GB"},
    "name": "thisIsATest",
}

cluster = Cluster(**cluster_options)

client = Client(cluster)

s3_file_handle_list = create_fileset(s3_file_uri_list)
GenericHandler().publish_cloud_optimised_fileset_batch(s3_file_handle_list)

[martindurant]

The scipy netCDF3 engine is certainly simplistic, as indeed is how the data is structured through the file. It may be the case, that just the stage of reading the coordinate values required reading the whole file - if this is in combination with failure to drop references, it would account for high memory usage and read bandwidth, but not big transfers to the cluster. I can confirm that pickle.dumps(f) of an S3File does not include the read buffer.

I am surprised that xarray would need to read the coordinates up front with "override" and "minimal" options, but you also have "decode_coords": True, and I don't really know what all these options do and how they combine.

You might want to consider kerchunk and/or virtualizarr to pre-scan all the files just ones and thereafter only have to refer to the manifest file. This should not be necessary, but you might find it useful in the long run anyway.

[lbesnard]

@martindurant thanks for your message. I'm not very familiar with Kerchunk. I tried using it over a years ago, but it wasn't working for my needs.
But those tools would still use the scipy engine and fsspec, so the issue would still be there.

Do you think it's a fsspec issue? Or should I reach out to scipy/xarray?

[martindurant]

But those tools would still use the scipy engine and fsspec

No! Kerchunk uses a modified version of the scipy engine (without data load) to do the initial indexing, and at read time you use the zarr engine for the actual load.

Do you think it's a fsspec issue? Or should I reach out to scipy/xarray?

My guess would be that it's down to the scipy implementation; but I don't think you will get any attention from them, since that code is very old and not designed for remote reads or cluster workflows. It would still be good to run memory profilers on your preprocess function to trace where the memory gets allocated, and to explicitly check the size of the generated dask graph before going further.

[lbesnard]

Thanks. I looked a bit into kerchunk. The new versions use zarr v3 which is still not production ready. I tried installing the older 2.18.7 version which uses zarr < v3.

As my knowledge of Kerchunk is more than limited, I'm probably not using it properly. But from what I've noticed:

• I can't use a preprocess function anymore as in xarray. So for example, it seems like I'm loosing the original filename as I use to have in the ds.encoding when files would be opened with open_mfdataset
• kerchunk has an NetCDF3ToZarr method and another one for h5. But apparently no method which could concatenate files together which are both NetCDF3 and 4 (unfortunately, this is what I have to deal with)
• it's not quite clear to me if Kerchunk uses dask automatically like in xarray.

I had a go at modifying my code to use kerchunk. Appart from my script failing at ds = xr.open_mfdataset(mappers, **open_mfdataset_params)

I still experience the same bejaviour, where data is sent to my local machine. But maybe this is not the correct way to do it

#!/usr/bin/env python3
from functools import partial

import s3fs
import xarray as xr
from coiled import Cluster
from dask.distributed import Client
import numpy as np
import numcodecs
import os

from dask import compute
from kerchunk.netCDF3 import NetCDF3ToZarr
from fsspec.mapping import FSMap


def _get_s3_path(file):
    path = getattr(file, "path", None) or getattr(file, "name", None)
    if not path:
        raise ValueError(f"File {file} has no path or name")
    if path.startswith("s3://"):
        return path
    bucket = getattr(file, "bucket", None)
    if bucket:
        # Strip bucket prefix if present in path
        if path.startswith(bucket + "/"):
            path = path[len(bucket) + 1 :]
        return f"s3://{bucket}/{path.lstrip('/')}"
    return path


def create_kerchunk_mappers(fileset):
    storage_options = {"anon": False}
    mappers = []
    for file in fileset:
        s3_uri = _get_s3_path(file)
        builder = NetCDF3ToZarr(s3_uri, storage_options=storage_options)
        mappers.append(builder.translate())
    return mappers


def create_fileset(s3_paths, s3_fs=None):
    """
    Create a fileset from S3 objects specified by a list of full S3 paths.

    Args:
        s3_paths (str or list[str]): Either a single full S3 path (e.g., 's3://bucket_name/object_key')
                                     or a list of full S3 paths.

    Returns:
        list[file-like object]: List of file-like objects representing each object in the fileset.
    """
    if s3_fs is None:
        s3_fs = s3fs.S3FileSystem(
            anon=True, default_fill_cache=False, default_cache_type="readahead"
        )

    if isinstance(s3_paths, str):
        s3_paths = [s3_paths]

    if not isinstance(s3_paths, list):
        raise ValueError("Invalid input format. Expecting either str or list[str].")

    # Create a fileset by opening each file
    fileset = [s3_fs.open(file) for file in s3_paths]

    return fileset


def preprocess_xarray(ds, dataset_config=None):
    append_dim = "TIME"
    dest_name = "filename"

    filename_placeholder = "UNKNOWN_FILENAME.nc"
    filename = filename_placeholder

    try:
        var = next(var for var in ds)  # get the first variable
        if hasattr(ds[var], "encoding") and "source" in ds[var].encoding:
            filename = os.path.basename(ds[var].encoding["source"])
        elif hasattr(ds, "encoding") and "source" in ds.encoding:
            filename = os.path.basename(ds.encoding["source"])

        print(f"Filename found in ds.encoding: {filename}")
    except Exception as e:
        print(
            f"Original filename not available in the xarray dataset: {e} - will default to use {filename_placeholder}"
        )
        filename = filename_placeholder

    length = ds.sizes[append_dim]
    string_array = np.full(
        length, filename, dtype=object
    )  # having da.full could break things when appending to existing zarr. need to investigate why. default to using numpy array instead

    ds[dest_name] = (append_dim, string_array)
    ds[dest_name].encoding = {
        "_FillValue": "",
        "dtype": object,
        "compressor": None,
        "filters": None,
        "object_codec": numcodecs.VLenUTF8(),  # crucial!
    }

    return ds


class GenericHandler:
    def _open_mfds(
        self, partial_preprocess, drop_vars_list, batch_files, engine="zarr"
    ):
        mappers = create_kerchunk_mappers(batch_files)
        # mappers = compute(*mappers)
        open_mfdataset_params = {
            "engine": engine,
            "parallel": True,
            "preprocess": partial_preprocess,
            "data_vars": "all",
            "concat_characters": True,
            "mask_and_scale": True,
            "decode_cf": True,
            "decode_times": True,
            "use_cftime": True,
            "decode_coords": True,
            "compat": "override",
            "coords": "minimal",
            "drop_variables": drop_vars_list,
            "backend_kwargs": {"consolidated": False},  # required for kerchunk
        }

        ds = xr.open_mfdataset(mappers, **open_mfdataset_params)

        # ds = ds.unify_chunks()
        # ds = ds.sortby("TIME")

        return ds

    def publish_cloud_optimised_fileset_batch(self, s3_file_uri_list):
        self.dataset_config = None
        drop_vars_list = ["wind_speed"]

        partial_preprocess = partial(
            preprocess_xarray, dataset_config=self.dataset_config
        )

        ds = self._open_mfds(
            partial_preprocess,
            drop_vars_list,
            s3_file_uri_list,
        )

    # next part of the code should do a ds.to_zarr()


prefix = "s3://imos-data/IMOS/SRS/OC/radiometer/VLHJ_Southern-Surveyor/2013/"
s3_file_uri_list = []
s3_file_uri_list += [
    prefix + "IMOS_SRS-OC_F_20130214T000001Z_VLHJ_FV01_DALEC_END-20130214T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130215T000000Z_VLHJ_FV01_DALEC_END-20130215T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130216T000000Z_VLHJ_FV01_DALEC_END-20130216T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130217T000002Z_VLHJ_FV01_DALEC_END-20130217T054432Z.nc",
    prefix + "IMOS_SRS-OC_F_20130218T213111Z_VLHJ_FV01_DALEC_END-20130218T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130219T000001Z_VLHJ_FV01_DALEC_END-20130219T054325Z.nc",
    prefix + "IMOS_SRS-OC_F_20130221T213337Z_VLHJ_FV01_DALEC_END-20130221T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130222T000004Z_VLHJ_FV01_DALEC_END-20130222T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130223T000002Z_VLHJ_FV01_DALEC_END-20130223T235951Z.nc",
    prefix + "IMOS_SRS-OC_F_20130224T000003Z_VLHJ_FV01_DALEC_END-20130224T230710Z.nc",
    prefix + "IMOS_SRS-OC_F_20130226T220957Z_VLHJ_FV01_DALEC_END-20130226T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130227T000001Z_VLHJ_FV01_DALEC_END-20130227T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130228T000000Z_VLHJ_FV01_DALEC_END-20130228T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130301T000002Z_VLHJ_FV01_DALEC_END-20130301T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130302T000001Z_VLHJ_FV01_DALEC_END-20130302T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130303T000000Z_VLHJ_FV01_DALEC_END-20130303T001757Z.nc",
    prefix + "IMOS_SRS-OC_F_20130410T001856Z_VLHJ_FV01_DALEC_END-20130410T061407Z.nc",
    prefix + "IMOS_SRS-OC_F_20130411T002135Z_VLHJ_FV01_DALEC_END-20130411T061533Z.nc",
    prefix + "IMOS_SRS-OC_F_20130412T002318Z_VLHJ_FV01_DALEC_END-20130412T061322Z.nc",
    prefix + "IMOS_SRS-OC_F_20130413T002039Z_VLHJ_FV01_DALEC_END-20130413T061708Z.nc",
    prefix + "IMOS_SRS-OC_F_20130414T001848Z_VLHJ_FV01_DALEC_END-20130414T061153Z.nc",
    prefix + "IMOS_SRS-OC_F_20130415T002210Z_VLHJ_FV01_DALEC_END-20130415T060559Z.nc",
    prefix + "IMOS_SRS-OC_F_20130416T001959Z_VLHJ_FV01_DALEC_END-20130416T055628Z.nc",
    prefix + "IMOS_SRS-OC_F_20130417T001719Z_VLHJ_FV01_DALEC_END-20130417T054750Z.nc",
    prefix + "IMOS_SRS-OC_F_20130418T001858Z_VLHJ_FV01_DALEC_END-20130419T040841Z.nc",
    prefix + "IMOS_SRS-OC_F_20130419T041404Z_VLHJ_FV01_DALEC_END-20130419T051434Z.nc",
    prefix + "IMOS_SRS-OC_F_20130420T002254Z_VLHJ_FV01_DALEC_END-20130420T044248Z.nc",
    prefix + "IMOS_SRS-OC_F_20130421T003058Z_VLHJ_FV01_DALEC_END-20130421T020501Z.nc",
    prefix + "IMOS_SRS-OC_F_20130429T005759Z_VLHJ_FV01_DALEC_END-20130429T031412Z.nc",
    prefix + "IMOS_SRS-OC_F_20130621T022830Z_VLHJ_FV01_DALEC_END-20130621T034358Z.nc",
    prefix + "IMOS_SRS-OC_F_20130622T015425Z_VLHJ_FV01_DALEC_END-20130622T040101Z.nc",
    prefix + "IMOS_SRS-OC_F_20130623T022616Z_VLHJ_FV01_DALEC_END-20130623T035247Z.nc",
    prefix + "IMOS_SRS-OC_F_20130624T023403Z_VLHJ_FV01_DALEC_END-20130624T042341Z.nc",
    prefix + "IMOS_SRS-OC_F_20130625T024217Z_VLHJ_FV01_DALEC_END-20130625T045224Z.nc",
    prefix + "IMOS_SRS-OC_F_20130626T025409Z_VLHJ_FV01_DALEC_END-20130626T050829Z.nc",
    prefix + "IMOS_SRS-OC_F_20130627T032159Z_VLHJ_FV01_DALEC_END-20130627T051355Z.nc",
    prefix + "IMOS_SRS-OC_F_20130728T003922Z_VLHJ_FV01_DALEC_END-20130728T071533Z.nc",
    prefix + "IMOS_SRS-OC_F_20130729T002107Z_VLHJ_FV01_DALEC_END-20130729T070924Z.nc",
    prefix + "IMOS_SRS-OC_F_20130730T000226Z_VLHJ_FV01_DALEC_END-20130730T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130731T000002Z_VLHJ_FV01_DALEC_END-20130731T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130801T000001Z_VLHJ_FV01_DALEC_END-20130801T235929Z.nc",
    prefix + "IMOS_SRS-OC_F_20130805T212533Z_VLHJ_FV01_DALEC_END-20130805T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130806T000002Z_VLHJ_FV01_DALEC_END-20130806T005357Z.nc",
    prefix + "IMOS_SRS-OC_F_20130807T042802Z_VLHJ_FV01_DALEC_END-20130807T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130808T000000Z_VLHJ_FV01_DALEC_END-20130808T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130809T000000Z_VLHJ_FV01_DALEC_END-20130809T063416Z.nc",
    prefix + "IMOS_SRS-OC_F_20130820T040551Z_VLHJ_FV01_DALEC_END-20130820T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130821T000002Z_VLHJ_FV01_DALEC_END-20130821T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130822T000001Z_VLHJ_FV01_DALEC_END-20130822T235955Z.nc",
    prefix + "IMOS_SRS-OC_F_20130823T000000Z_VLHJ_FV01_DALEC_END-20130823T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130824T000001Z_VLHJ_FV01_DALEC_END-20130824T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130825T000002Z_VLHJ_FV01_DALEC_END-20130825T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130826T000004Z_VLHJ_FV01_DALEC_END-20130826T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130827T000003Z_VLHJ_FV01_DALEC_END-20130827T211558Z.nc",
    prefix + "IMOS_SRS-OC_F_20130828T204304Z_VLHJ_FV01_DALEC_END-20130828T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130829T000000Z_VLHJ_FV01_DALEC_END-20130829T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130830T000003Z_VLHJ_FV01_DALEC_END-20130830T235956Z.nc",
    prefix + "IMOS_SRS-OC_F_20130831T000001Z_VLHJ_FV01_DALEC_END-20130831T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130901T000000Z_VLHJ_FV01_DALEC_END-20130901T235954Z.nc",
    prefix + "IMOS_SRS-OC_F_20130902T000000Z_VLHJ_FV01_DALEC_END-20130902T034340Z.nc",
    prefix + "IMOS_SRS-OC_F_20130927T201704Z_VLHJ_FV01_DALEC_END-20130927T235956Z.nc",
    prefix + "IMOS_SRS-OC_F_20130928T000001Z_VLHJ_FV01_DALEC_END-20130928T235955Z.nc",
    prefix + "IMOS_SRS-OC_F_20130930T203535Z_VLHJ_FV01_DALEC_END-20130930T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20131001T000000Z_VLHJ_FV01_DALEC_END-20131001T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20131002T000003Z_VLHJ_FV01_DALEC_END-20131002T034135Z.nc",
]


cluster_options = {
    "n_workers": [1, 120],
    "scheduler_vm_types": "t3.2xlarge",
    "worker_vm_types": "t3.2xlarge",
    "allow_ingress_from": "me",
    "compute_purchase_option": "spot_with_fallback",
    "worker_options": {"nthreads": 8, "memory_limit": "32GB"},
    "name": "thisIsATest",
}

cluster = Cluster(**cluster_options)

client = Client(cluster)

s3_file_handle_list = create_fileset(s3_file_uri_list)
GenericHandler().publish_cloud_optimised_fileset_batch(s3_file_handle_list)

[lbesnard]

I tried a different approach with kerchunk but the processing still happens locally instead of happening in the cluster:

import os

import fsspec
import numcodecs
import numpy as np
import ujson
import xarray as xr
from coiled import Cluster
from dask.distributed import Client
from kerchunk.combine import MultiZarrToZarr
from kerchunk.netCDF3 import NetCDF3ToZarr
from tqdm import tqdm

prefix = "s3://imos-data/IMOS/SRS/OC/radiometer/VLHJ_Southern-Surveyor/2013/"
s3_file_uri_list = []
s3_file_uri_list += [
    prefix + "IMOS_SRS-OC_F_20130214T000001Z_VLHJ_FV01_DALEC_END-20130214T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130215T000000Z_VLHJ_FV01_DALEC_END-20130215T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130216T000000Z_VLHJ_FV01_DALEC_END-20130216T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130217T000002Z_VLHJ_FV01_DALEC_END-20130217T054432Z.nc",
    prefix + "IMOS_SRS-OC_F_20130218T213111Z_VLHJ_FV01_DALEC_END-20130218T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130219T000001Z_VLHJ_FV01_DALEC_END-20130219T054325Z.nc",
    prefix + "IMOS_SRS-OC_F_20130221T213337Z_VLHJ_FV01_DALEC_END-20130221T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130222T000004Z_VLHJ_FV01_DALEC_END-20130222T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130223T000002Z_VLHJ_FV01_DALEC_END-20130223T235951Z.nc",
    prefix + "IMOS_SRS-OC_F_20130224T000003Z_VLHJ_FV01_DALEC_END-20130224T230710Z.nc",
    prefix + "IMOS_SRS-OC_F_20130226T220957Z_VLHJ_FV01_DALEC_END-20130226T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130227T000001Z_VLHJ_FV01_DALEC_END-20130227T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130228T000000Z_VLHJ_FV01_DALEC_END-20130228T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130301T000002Z_VLHJ_FV01_DALEC_END-20130301T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130302T000001Z_VLHJ_FV01_DALEC_END-20130302T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130303T000000Z_VLHJ_FV01_DALEC_END-20130303T001757Z.nc",
    prefix + "IMOS_SRS-OC_F_20130410T001856Z_VLHJ_FV01_DALEC_END-20130410T061407Z.nc",
    prefix + "IMOS_SRS-OC_F_20130411T002135Z_VLHJ_FV01_DALEC_END-20130411T061533Z.nc",
    prefix + "IMOS_SRS-OC_F_20130412T002318Z_VLHJ_FV01_DALEC_END-20130412T061322Z.nc",
    prefix + "IMOS_SRS-OC_F_20130413T002039Z_VLHJ_FV01_DALEC_END-20130413T061708Z.nc",
    prefix + "IMOS_SRS-OC_F_20130414T001848Z_VLHJ_FV01_DALEC_END-20130414T061153Z.nc",
    prefix + "IMOS_SRS-OC_F_20130415T002210Z_VLHJ_FV01_DALEC_END-20130415T060559Z.nc",
    prefix + "IMOS_SRS-OC_F_20130416T001959Z_VLHJ_FV01_DALEC_END-20130416T055628Z.nc",
    prefix + "IMOS_SRS-OC_F_20130417T001719Z_VLHJ_FV01_DALEC_END-20130417T054750Z.nc",
    prefix + "IMOS_SRS-OC_F_20130418T001858Z_VLHJ_FV01_DALEC_END-20130419T040841Z.nc",
    prefix + "IMOS_SRS-OC_F_20130419T041404Z_VLHJ_FV01_DALEC_END-20130419T051434Z.nc",
    prefix + "IMOS_SRS-OC_F_20130420T002254Z_VLHJ_FV01_DALEC_END-20130420T044248Z.nc",
    prefix + "IMOS_SRS-OC_F_20130421T003058Z_VLHJ_FV01_DALEC_END-20130421T020501Z.nc",
    prefix + "IMOS_SRS-OC_F_20130429T005759Z_VLHJ_FV01_DALEC_END-20130429T031412Z.nc",
    prefix + "IMOS_SRS-OC_F_20130621T022830Z_VLHJ_FV01_DALEC_END-20130621T034358Z.nc",
    prefix + "IMOS_SRS-OC_F_20130622T015425Z_VLHJ_FV01_DALEC_END-20130622T040101Z.nc",
    prefix + "IMOS_SRS-OC_F_20130623T022616Z_VLHJ_FV01_DALEC_END-20130623T035247Z.nc",
    prefix + "IMOS_SRS-OC_F_20130624T023403Z_VLHJ_FV01_DALEC_END-20130624T042341Z.nc",
    prefix + "IMOS_SRS-OC_F_20130625T024217Z_VLHJ_FV01_DALEC_END-20130625T045224Z.nc",
    prefix + "IMOS_SRS-OC_F_20130626T025409Z_VLHJ_FV01_DALEC_END-20130626T050829Z.nc",
    prefix + "IMOS_SRS-OC_F_20130627T032159Z_VLHJ_FV01_DALEC_END-20130627T051355Z.nc",
    prefix + "IMOS_SRS-OC_F_20130728T003922Z_VLHJ_FV01_DALEC_END-20130728T071533Z.nc",
    prefix + "IMOS_SRS-OC_F_20130729T002107Z_VLHJ_FV01_DALEC_END-20130729T070924Z.nc",
    prefix + "IMOS_SRS-OC_F_20130730T000226Z_VLHJ_FV01_DALEC_END-20130730T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130731T000002Z_VLHJ_FV01_DALEC_END-20130731T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130801T000001Z_VLHJ_FV01_DALEC_END-20130801T235929Z.nc",
    prefix + "IMOS_SRS-OC_F_20130805T212533Z_VLHJ_FV01_DALEC_END-20130805T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130806T000002Z_VLHJ_FV01_DALEC_END-20130806T005357Z.nc",
    prefix + "IMOS_SRS-OC_F_20130807T042802Z_VLHJ_FV01_DALEC_END-20130807T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130808T000000Z_VLHJ_FV01_DALEC_END-20130808T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130809T000000Z_VLHJ_FV01_DALEC_END-20130809T063416Z.nc",
    prefix + "IMOS_SRS-OC_F_20130820T040551Z_VLHJ_FV01_DALEC_END-20130820T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130821T000002Z_VLHJ_FV01_DALEC_END-20130821T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130822T000001Z_VLHJ_FV01_DALEC_END-20130822T235955Z.nc",
    prefix + "IMOS_SRS-OC_F_20130823T000000Z_VLHJ_FV01_DALEC_END-20130823T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130824T000001Z_VLHJ_FV01_DALEC_END-20130824T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130825T000002Z_VLHJ_FV01_DALEC_END-20130825T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20130826T000004Z_VLHJ_FV01_DALEC_END-20130826T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130827T000003Z_VLHJ_FV01_DALEC_END-20130827T211558Z.nc",
    prefix + "IMOS_SRS-OC_F_20130828T204304Z_VLHJ_FV01_DALEC_END-20130828T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130829T000000Z_VLHJ_FV01_DALEC_END-20130829T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20130830T000003Z_VLHJ_FV01_DALEC_END-20130830T235956Z.nc",
    prefix + "IMOS_SRS-OC_F_20130831T000001Z_VLHJ_FV01_DALEC_END-20130831T235959Z.nc",
    prefix + "IMOS_SRS-OC_F_20130901T000000Z_VLHJ_FV01_DALEC_END-20130901T235954Z.nc",
    prefix + "IMOS_SRS-OC_F_20130902T000000Z_VLHJ_FV01_DALEC_END-20130902T034340Z.nc",
    prefix + "IMOS_SRS-OC_F_20130927T201704Z_VLHJ_FV01_DALEC_END-20130927T235956Z.nc",
    prefix + "IMOS_SRS-OC_F_20130928T000001Z_VLHJ_FV01_DALEC_END-20130928T235955Z.nc",
    prefix + "IMOS_SRS-OC_F_20130930T203535Z_VLHJ_FV01_DALEC_END-20130930T235958Z.nc",
    prefix + "IMOS_SRS-OC_F_20131001T000000Z_VLHJ_FV01_DALEC_END-20131001T235957Z.nc",
    prefix + "IMOS_SRS-OC_F_20131002T000003Z_VLHJ_FV01_DALEC_END-20131002T034135Z.nc",
]


cluster_options = {
    "n_workers": [1, 120],
    "scheduler_vm_types": "t3.2xlarge",
    "worker_vm_types": "t3.2xlarge",
    "allow_ingress_from": "me",
    "compute_purchase_option": "spot_with_fallback",
    "worker_options": {"nthreads": 8, "memory_limit": "32GB"},
    "name": "thisIsATest",
}

cluster = Cluster(**cluster_options)

client = Client(cluster)


def preprocess_xarray(ds, dataset_config=None):
    append_dim = "TIME"
    dest_name = "filename"

    filename_placeholder = "UNKNOWN_FILENAME.nc"
    filename = filename_placeholder

    try:
        var = next(var for var in ds)  # get the first variable
        if hasattr(ds[var], "encoding") and "source" in ds[var].encoding:
            filename = os.path.basename(ds[var].encoding["source"])
        elif hasattr(ds, "encoding") and "source" in ds.encoding:
            filename = os.path.basename(ds.encoding["source"])

        print(f"Filename found in ds.encoding: {filename}")
    except Exception as e:
        print(
            f"Original filename not available in the xarray dataset: {e} - will default to use {filename_placeholder}"
        )
        filename = filename_placeholder

    length = ds.sizes[append_dim]
    string_array = np.full(
        length, filename, dtype=object
    )  # having da.full could break things when appending to existing zarr. need to investigate why. default to using numpy array instead

    ds[dest_name] = (append_dim, string_array)
    ds[dest_name].encoding = {
        "_FillValue": "",
        "dtype": object,
        "compressor": None,
        "filters": None,
        "object_codec": numcodecs.VLenUTF8(),  # crucial!
    }

    return ds


references = []

fs = fsspec.filesystem("s3", anon=True)

for url in tqdm(s3_file_uri_list, desc="Kerchunking files"):
    h = NetCDF3ToZarr(url, storage_options={"anon": True})
    single_ref = h.translate()
    references.append(single_ref)

#  The next bit fails with chunk size
#    raise ValueError(
#ValueError: Found chunk size mismatch:
#                       at prefix relaz in iteration 1 (file None)
#                        new chunk: [9563]
# mzz = MultiZarrToZarr(
#     references,
#     remote_protocol="s3",
#     remote_options={"anon": True},
#     concat_dims=["TIME"],
#     identical_dims=["LATITUDE", "LONGITUDE"],
# )
#
# combined_ref = mzz.translate()
#
# with open("combined_reference.json", "w") as f:
#     ujson.dump(combined_ref, f)
#
#
# mapper = fsspec.get_mapper(
#     "reference://", fo=combined_ref, remote_protocol="s3", remote_options={"anon": True}
# )
#
# ds = xr.open_dataset(mapper, engine="zarr", backend_kwargs={"consolidated": False})
# ds = preprocess_xarray(ds)

I've found it quite challenging to get everything working smoothly. There aren't many comprehensive tutorials or examples for this specific use case (which at the end of the day is pretty basic), especially when incorporating tools like fsspec, kerchunk, and Dask in a distributed environment.

As a user, I find the ecosystem a bit overwhelming at times with many similar tools (like Icechunk, VirtualZarr, Pangeo Forge, etc.) emerging frequently, each targeting a slightly different aspect of the problem. It feels like there's a growing fragmentation in the tooling landscape, and it can be difficult to figure out which pieces are stable, recommended or compatible.

[martindurant]

A comment on the latter...

xarray-zarr-kerchunk-fsspec were working and stable under zarr2, although kerchunk lacked some generalization features (a full set of combine scenarios) and zarr2 was not capable of adding features (like variable chunk sizes). The indexing/merging stage was never quite easy (each data set is very unique!), but the end-user side was solid.

With the coming of zarr3, a whole lot of work has gone into other projects like vzarr and icechunk, which were not necessarily dependent on one-another, but were contemporaneous. This followed an extremely extended planning period for zarr3, and the release this year has taken a long time to settle across the various tools. Since vzarr and icechunk wish to take over some of the functionality formerly only in kerchunk, and other libraries (obstore for fsspec) and also in the mix, there's not much that I can do from my corner of things.
(Indeed, pangeo-forge has also been lost in the mix)