dask / dask/distributed

Spill to disk is inefficient for small byte-values

Open
#525 22 comments 0 reactions 0 assignees View on GitHub
enhancement
Dominant language
Python
Stars
1.7k
Forks
778
Avg merge
2h 50m
Merged PRs (30d)
3

Description

As of #485 workers can spill excess data to disk. They currently evict tasks from a fixed pool of memory to disk with a simple LRU policy. Task results are stored on disk as single files. This is currently done using the composable MutableMappings in a tiny project, [zict](https://github.com/mrocklin/zict).

``` python
if memory_limit:
from zict import Func, Buffer, File
path = os.path.join(self.local_dir, 'storage')
memory = dict()
disk = File(path)
disk = Func(dumps_to_disk, loads_from_disk, disk)
self.data = Buffer(memory, disk, int(float(memory_limit)), weight) # LRU from memory to disk
else:
self.data = {}
```

This policy is decent for large results (larger than say, a megabyte) but will slow down when trying to save many small results due to file system overhead. This case specifically occurs when performing shuffles, such as with dask.array rechunkings, or dask.bag/dataframe joins, where we have possibly millions of very small results. Saving thousands of tiny files is _very_ slow on most file systems. [See blogpost](http://matthewrocklin.com/blog/work/2015/12/29/disk-bandwidth).

Single-machine key-value store databases provide decently efficient random write and read access to small values on disk. There are _many_ solutions here. I'm somewhat partial to embedded databases like LevelDB (which now has many more modern competitors) that run inside the same process because they don't require significant setup, run in a well contained way, and are easy to use.

So one solution here would be to optionally replace the `zict.File(MutableMapping)` that uses the filesystem with a MutableMapping that was backed by some more efficient database. This would require learning a bit about different key-value stores and choosing a good one based on the many-small writes/reads workloads that dask.distributed might push onto it, finding a way to combine it with the LRU scheme (or some other scheme) used currently (this is easy if you can make it follow a `MutableMapping` interface), and then hooking up the necessary keywords to allow users to access this option easily.

Key-value stores may be bad for large-byte values. We might want a three-MutableMapping solution. One for in memory, one for large-on-disk (zict.File) and one for small-on-disk (some on-disk database).

Some considerations:
1. We need to be able to specify/restrict the amount of memory that the database uses internally
2. I would love to see more MutableMapping added to Zict, which I should move to the dask org if anything like this happens.
3. ...

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.