Genetics data IO performance stats/doc #437
Comments
|
Some doc for fsspec:
This is a partial output of the method profile on the code above (https://github.com/pystatgen/sgkit/issues/437#issue-784385179): Unsurprising we get numpy methods, what is interesting and can't be seen here, are the sporadic spikes in CPU usage, looking at the network IO: There are also spikes of network IO, which could suggest that we should look into smoothing out the network buffering. This reminded me about some stats I once stumbled upon in Beam codebase (here): So Beam is using 16MiB read buffer. But it's also using Google's storage API client (python-storage), whilst fsspec calls GCS API directly via fsspec implementation of GCS gcsfs and there definitely are implementation differences between those two and thus the consequences of buffer size might be different. |
|
In the original issue: I have tried fsspec block size of 16MiB, but the results look very similar to the 5MiB block size, which was somewhat suspicious, and turns out that the zarr data ("gs://foobar/data.zarr") is saved as a large number of small files, for example for the
From S3 best practices.
From cloud performant reading of netcdf4 hdf5 data using the zarr library. Interesting/related issue "File Chunk Store" zarr-python#556. To validate the impact of the chunk size, let's rechunked the Performance graph:
There is a cost to it tho, visible in the memory usage, notice that compressed chunk of 17MiB translates to 2GiB in memory (vs 2MiB -> 230MiB), this bumps the memory usage roughly from 6GB to 40GB. We could look into investigating the cost of reading zarr data in smaller chunks (then it was written as). This is a profile of the computation with zarr chunks
Notice:
Say we have a random float zarr array shape We retrieve a single element: zs = zarr.open("/tmp/zarr", mode="r")
print(zs.foo[0, 0])Zarr will:
So overall the lesson from this exercise is that - slicing zarr in a way that doesn't align with the native chunks is inefficient as a common step in pipelines. Zarr issues for partial chunk reads:
Highlights:
|
|
Some experiments that incorporate xarray + zarr + dask: TLDR
Long storySay we have a dataset: fsmap = fsspec.get_mapper("/tmp/zarr")
ar = np.random.random((10_000, 100_000))
xar = xr.Dataset(data_vars={"foo": (("x", "y"), ar)})
xar.foo.encoding["chunks"] = (10_000, 10_000)
xar.to_zarr(fsmap)This gives us a zarr store with one 2d array stored as 10 chunks (10_000, 10_000). Now, say we read this back via: xar_back = xr.open_zarr(fsmap)Dask graph looks like this:
and an example of one zarr task definition: Notes:
The question now is - how can we force async zarr multiget via dask.
xar_back = xr.open_zarr(fsmap).chunk({"y": 20_000, "x": 10_000})This won't work: Notes:
xar_back = xr.open_zarr(fsmap, chunks={"y": 20_000, "x": 10_000})we get:
and an example of one zarr task definition: Notes:
Another interesting question is: how would dask react to chunking that doesn't align with zarr native chunks: xar_back = xr.open_zarr(fsmap, chunks={"y": 12_500, "x": 10_000})Notice that When we run the code above, xarray tells us:
Let's see the graph:
and an example of two zarr task definitions: Notes:
|
Experiments with simplified computation and random data.We generate reasonably large float array 1e5 x 1e5 and save it in large chunks (400MB) and small chunks (4MB). Then run 3 experiments (many times 30 - 100):
In the (X/Y/Z) notation, X stands for chunk size at rest time (storage), Y is read chunk size (the size zarr will use to read data from storage, at this point the data is already in memory), Z stands for chunk size used for computation. In each experiment we multiple the array by a scalar, and use 4MB chunking for this computation (this is to imitate that our problem requires certain chunk size that doesn't necessarily correspond to the zarr chunk size). We run the computations many times and measure time/memory/network. All data is stored on GCS (in the same region as the VM). Dask cluster setup:We are using distributed process pool based cluster on a single 16 core VM, with 16 workers/processes with 1 thread each. This is to avoid any GIL issues, and since the computation is embarrassingly parallel, unless there is some job stilling happening there is very little communication between workers. We run 2 rounds of computation to warm up the cluster. I have also tried a thread pool based cluster but was seeing GIL impact. Results
Notes:
And next a macro (much less accurate than the data used for the network histogram above) view of the CPU/IO (Network)/Memory usage for the cases (400/400/4) and (4/400/4). First we perform many 400/400/4 from 3pm until around 4:30, and then switch to 4/400/4 until the end. Notes:
Conclusion: this validates that larger chunks result in better throughput even when compared with concurrent read of many small chunks. But reading large chunks (or many small chunks) comes at the cost of memory (as mentioned in the previous comments). CodeCode: generate the datadef create_data(path, chunks):
fs_map = gcs.get_mapper(path)
ar_dsk = da.random.random(size=(100_000, 100_000), chunks=chunks).astype(np.float32)
ar_xr = xr.Dataset(data_vars=(dict(foo=(("x", "y"), ar_dsk))))
with ProgressBar(), ResourceProfiler() as rprof:
ar_xr.to_zarr(fs_map, mode="w")Code: computationsdef large_chunks(path):
# 16MB block size
gcs = gcsfs.GCSFileSystem(block_size=16 * 1024 * 1024)
fs_map = gcs.get_mapper(path)
# use zarr chunking
ar_xr = xr.open_zarr(fs_map)
# rechunk/split to smaller chunks
ar_xr["foo"] = ar_xr.foo.chunk(dict(x=1_000, y=1_000))
# Simple computation on part of the matrix
dat = (ar_xr.foo * 2.0).data.persist()
wait(dat)
del dat
def small_chunks(path):
# default block size of 5MB
gcs = gcsfs.GCSFileSystem()
fs_map = gcs.get_mapper(path)
# read in many zarr chunks at the same time
ar_xr = xr.open_zarr(fs_map, chunks=dict(x=10_000, y=10_000))
# rechunk/split to smaller chunks
ar_xr["foo"] = ar_xr.foo.chunk(dict(x=1_000, y=1_000))
# Simple computation on part of the matrix
dat = (ar_xr.foo * 2.0).data.persist()
wait(dat)
del dat
def native_chunks(path):
# default block size of 5MB
gcs = gcsfs.GCSFileSystem()
fs_map = gcs.get_mapper(path)
# use zarr chunking
ar_xr = xr.open_zarr(fs_map)
# Simple computation on part of the matrix
dat = (ar_xr.foo * 2.0).data.persist()
wait(dat)
del dat |
|
GCS stats:
|
|
TLDR thus far:
Another discussion that came up is the memory impact on the performance, given that larger chunk will require more memory. afaiu by default in a dask cluster:
In GCP, default VMs are divided into classes of machines like “standard” or “highmem”, in each class the memory to CPU ratio is constant, for |
|
A lot to digest here, thanks for the great work @ravwojdyla! |
ravwojdyla
added
documentation
|
And a way to connect these with #390, I personally would be very curious to see the #390 GWAS benchmark with:
|
|
Based on the performance tests done above, here are some high level guidelines for dask performance experiments (this is a starting point, we might find a better home for this later, and potentially have someone from Dask review them):
def get_dask_cluster(n_workers=1, threads_per_worker=None):
dk.config.set({"distributed.worker.memory.terminate": False})
workers_kwargs = {"memory_target_fraction": False,
"memory_spill_fraction": False,
"memory_pause_fraction": .9}
return Client(n_workers=n_workers,
threads_per_worker=threads_per_worker,
**workers_kwargs)
TODO:
|
This is a dump of some of the performance experiments. It's part of a larger issue of performance setup and best practices for dask/sgkit and genetic data. The goal is to share the findings and continue the discussion.
Where not otherwise stated, the test machine is a GCE VM, 16 cores and 64GB of memory, 400 SPD. Dask cluster is a single node process based. If the data is read from GCS, the bucket is in the same region as the VM:
Specs/libs
The issue with suboptimal saturation was originally reported for this code:
With local input, performance graph:
It's pretty clear the cores are well saturated. I also measure GIL, GIL was held for 13% of time and waited on for 2.1%, with each worker thread (16 threads) holding it for 0.7% and waiting for 0.1% of time.
For GCS input (via fsspec):
GIL summary: GIL was held for 18% of time and waited on for 3.8%, with each worker thread (16 threads) holding it for 0.6% and waiting for 0.2% of time, with one thread holding GIL for 6.5% and waiting 1.6% time.
It's clear that the CPU usage is lower, and not fully saturated, GIL wait time is a bit up (with a concerning spike in one thread). With remote/fsspec input, we have the overhead of data decryption and potential network IO overhead (tho it doesn't seem like we hit network limits).
The text was updated successfully, but these errors were encountered: