Add support for dask dataframes

[krfricke]

Closes #92

Note that the locality-aware scheduling can currently not be tested reliably. This is due to 2 reasons:

1. We cannot guarantee that dask partitions end up living on a specific node (to set an initial state for redistribution)
2. We cannot guarantee that tasks that determine the dask partition node are co-scheduled with the respective partition.

Once the ray.state.objects() API or something similar comes back, the second option should be easy to enable, and the first option will be easy to confirm.

[krfricke]

Currently it seems that the problem lies with 1) - ray memory shows that each partition lives on the head node after calling persist(). I'll try to find a solution to that.

[richardliaw]

will this materialize the full distributed dataframe?

[krfricke]

Yes. However, this part of the code concerns the local loading - i.e. loading of either a partition or centralized loading, where access to the whole dataframe is assumed anyway. This is the part where we end up with a pandas dataframe.

[richardliaw]

hmm, why would it be misinterpreted?

[krfricke]

In load_data(), we are doing this check:

        if indices is not None and len(indices) > 0 and isinstance(
                indices[0], Tuple):

Usually, numerical indices would indicate the row numbers a worker should load. This is also valid for Dask (e.g. for centralized loading). However, in this case the numbers should indicate the partition numbers. To not confuse these with row indices, we use a different type. IMO tuples are a good choice because they are immutable and should produce very few overhead.

Note that in Modin we use Ray ObjectIDs instead, but we don't have these for Dask at this point in the code.

[richardliaw]

Generally looks great! The main question i have is about some of the implementation details.