pydata / pydata/xarray

open_mfdataset failed to open tarfile filestream when it locates in the context of dask.distributed Client

Open
#5,043 1 comment 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

topic-backends topic-dask
Dominant language
Python
Stars
4.2k
Forks
1.4k
Avg merge
2d 15h
Merged PRs (30d)
14

Description

Recently, i use open_mfdataset to open a local tar.gz file of multiple netcdf files,
it failed to open it and raise a distributed.scheduler.KilledWorker: Error and
TypeError: cannot serialize 'ExFileObject' object .

My code is like following,

import tarfile
from dask.distributed import Client

client = Client()

tar = tarfile.open(my_multiple_netcdf_tar_gz_file)
flist = [tar.extractfile(member) for member in tar.getmembers()]

ds = xr.open_mfdataset(flist)

# This line will raise Exception
print(ds.MyNcVar.values)

# ....
# blah blah my other client calcualation codes
# ....

client.close()

In above code, the elements of variable flist will be type of ExFileObject,
which can't be serialized to distributed.Client cluster and
therefore will result in the failure of open_mfdataset .

The reason is xr.open_mfdataset auto convert chunks=None to {} ,
which will force the method xr.open_dataset to use dask.

We can see in this line of open_mfdataset ,

https://github.com/pydata/xarray/blob/37fe5441c8a2fb981f2c50b8379d7d4f8492ae19/xarray/backends/api.py#L897

    # Notes this line will force chunks=None into chunks={} and result in the involvement of dask
    open_kwargs = dict(engine=engine, chunks=chunks or {}, **kwargs)

    if parallel:
        import dask

        # wrap the open_dataset, getattr, and preprocess with delayed
        open_ = dask.delayed(open_dataset)
        getattr_ = dask.delayed(getattr)
        if preprocess is not None:
            preprocess = dask.delayed(preprocess)
    else:
        open_ = open_dataset
        getattr_ = getattr

    datasets = [open_(p, **open_kwargs) for p in paths]
    closers = [getattr_(ds, "_close") for ds in datasets]

Even if i set the chunks=None, it will be a error cause the chunks always not be None
when it is passed into open_dataset .

I think maybe we can keep the chunks value and if anyone want change it,
he or she can set it to {} or any other values as they want ?

    open_kwargs = dict(engine=engine, chunks=chunks, **kwargs)

Or may you have a better solution for my problem ?

Also, Thank You for your great jobs on this excellent package.

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start in xarray/backends/api.py around the open_mfdataset open_kwargs construction referenced by the issue. Reproduce the tarfile ExFileObject case with a dask.distributed Client, then verify that preserving chunks=None avoids serialization failure while explicit chunk values still work.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
data, distributed-systems
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Stale
Clarity
Clearly specified
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.