distributed out-of-core computation with chunks

[tanmaykm]

This implements a distributed out-of-core model for ALS.

The key component used here is a ChunkedFile. It's a file split into chunks, with metadata describing the range of keys held in each chunk. Chunks can be loaded independently. Loaded data is weakly referenced, so they get garbage collected on memory pressure. Based on available memory, some references are kept in a LRU cache with a configurable limit.

The current implementation stores two types of data structures as chunks, both of which are memory mapped:

• SparseMatChunks: used to represent the input ratings. Methods mmap_csc_save and mmap_csc_load handle the file format. Since input data is relatively constant, and it's impractical to do the same ETL process on large inputs everytime, they should be precomputed and stored as chunks. A transpose of the ratings matrix is also pre-computed. The data must be clean, empty items and users removed, and user/item id mappings must be handled separately.
• DenseMatChunks: used to represent the model (U and P). These are created at run time. Chunk size is calculated automatically to fit one chunk into a 128MB file.

Concrete types for Model and Inputs for distributed memory mode implement the abstractions introduced by #21.

• DistModel uses memory mapped dense matrix chunks
• DistInputs uses memory mapped sparse matrix chunks

I have tried this with the movielens and last.fm datasets. Both seem to be running correctly, with much less memory pressure. With this, I am able to complete a run on the last.fm dataset on my laptop which otherwise runs out of memory. This mode is much slower than shared memory mode though, as expected.

Way forward:

• test this with larger datasets
• performance comparisons
• use parquet format, instead of a custom format
• should be able to work on HDFS

[tanmaykm]

Some performance comparisons of chunks with shared memory parallelism mode on two types of machines:

• laptop, with 16GB RAM, 4 cores (x2 hyperthreaded)
• julia.mit.edu machine, with 1T RAM, 40 cores (x2 hyperthreaded, slower than the laptop cores)

Data sets used:

• D: last.fm dataset (users: 148,111, items: 1,568,126, 24,296,858 observations, ~1GB mmapped size, split across 20 chunks)
• 4D: 4 x last.fm dataset (generated by duplicating rows and columns of D)
• 25D: 25 x last.fm dataset

All tests were done with:

• 8 workers
• 20 iterations
• chunk cache: 10 for each file

The laptop memory usage was maxed out with the smallest dataset in shared memory mode, and could not complete the rmse validations. It however ran to completion with chunks, even with the larger datasets.

machine	dataset	mode	time (sec)
julia.mit.edu	25D	chunks	37658
julia.mit.edu	D	chunks	1646
julia.mit.edu	D	shmem (pmap)	5770
julia.mit.edu	D	shmem (@parallel)	629
laptop	4D	chunks	2951
laptop	D	chunks	875
laptop	D	shmem (pmap)	2220
laptop	D	shmem (@parallel)	320

[tanmaykm]

Will take up HDFS and Parquet interfaces as separate PRs.

[ViralBShah]

@amitmurthy - Huge gap between pmap and @parallel. We know why but need to find a good solution

[ViralBShah]

Cc @andreasnoack

[ViralBShah]

cc @simonbyrne

I also think some of these notes should go into the README.