open_mfdataset failed to open tarfile filestream when it locates in the context of dask.distributed Client
Nobody has claimed this yet.
- 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 ,
# 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
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- 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