Repository: modin-project/modin
Pull request: Add Dask futures implementation
<!--
Thank you for your contribution!
-->

## What do these changes do?

<!-- Please give a short brief about these changes. -->

## Related issue number
Resolves #657 

<!-- Are there any issues opened that will be resolved by merging this change? -->

- [x] passes `flake8 modin`
- [x] passes `black --check modin`
- [ ] tests added and passing


Problem statement:
Windows support
Ray does not support Windows. It would be good to publish wheels that don't require Ray for Windows users. This is also related to publishing to conda-forge.

We can add an engine that runs strictly off of Python Multiprocessing for starters. It should be  relatively straight-forward to add.

Marking as Help Wanted, feel free to reach out about required implementation details.

Hints from issue discussion:
Hi @devin-petersohn. I started working on a simply multiprocess engine.
My approach is quite naive. In the partition implementation I simply send the work to a different process. Similar to what Ray does with a remote task. 
Had some serialization issues so for now using dill instead of pickle.
I have run it on the following example and it indeed works and opens new python processes for each partition
```python
import modin.pandas as pd
import numpy as np

data = np.random.randint(0, 7, size=(10000, 100))
df = pd.DataFrame(data)
print(df)
print(df.count())
```
My code is far from complete, hence, I am not opening a PR just yet (unless we want to put it in WIP).
Following is my partition class. 
Does this approach makes sense? feedbacks would be appreciated 
```python
import pandas
from multiprocessing import Pool

from modin.data_management.utils import length_fn_pandas, width_fn_pandas
from modin.engines.base.frame.partition import BaseFramePartition


def func_wrapper(call_queue, data):
    import dill

    def deserialize(obj):
        if isinstance(obj, bytes):
            return dill.loads(obj)
        return obj

    for func, kwargs in call_queue:
        func = deserialize(func)
        kwargs = deserialize(kwargs)
        data = func(data, **kwargs)

    return data


def deploy_mp(call_queue, data):
    pid = pool.starmap_async(func_wrapper, [(call_queue, data), ])
    return pid


def put(obj):
    return obj


def put_mp(obj):
    pid = pool.map_async(put, (obj,))
    return pid


pool = Pool()


class PandasOnMultiprocessFramePartition(BaseFramePartition):
    """This abstract class holds the data and metadata for a single partition.
        The methods required for implementing this abstract class are listed in
        the section immediately following this.

        The API exposed by the children of this object is used in
        `BaseFrameManager`.

        Note: These objects are treated as immutable by `BaseFrameManager`
        subclasses. There is no logic for updating inplace.
    """

    def __init__(self, pid, length=None, width=None, call_queue=[]):
        self.pid = pid
        self.call_queue = call_queue
        self._length_cache = length
        self._width_cache = width

    def get(self):
        """Flushes the call_queue and returns the data.

        Note: Since this object is a simple wrapper, just return the data.

        Returns:
            The object that was `put`.
        """
        self.drain_call_queue()

        # blocking operation
        df = self.pid.get()[0]
        return df

    def apply(self, func, **kwargs):
        """Apply some callable function to the data in this partition.

        Note: It is up to the implementation how kwargs are handled. They are
            an important part of many implementations. As of right now, they
            are not serialized.

        Args:
            func: The lambda to apply (may already be correctly formatted)

        Returns:
             A new `BaseFramePartition` containing the object that has had `func`
             applied to it.
        """

        call_queue = self.call_queue + [(func, kwargs)]
        self.call_queue = []

        pid = deploy_mp(call_queue, self.get().copy())
        return PandasOnMultiprocessFramePartition(pid)

    def add_to_apply_calls(self, func, **kwargs):
        return PandasOnMultiprocessFramePartition(
            self.pid, call_queue=self.call_queue + [(func, kwargs)]
        )

    def drain_call_queue(self):
        if len(self.call_queue) == 0:
            return
        self.apply(lambda x: x)

    def mask(self, row_indices=None, col_indices=None):
        new_obj = self.add_to_apply_calls(
            lambda df: pandas.DataFrame(df.iloc[row_indices, col_indices])
        )
        new_obj._length_cache, new_obj._width_cache = len(row_indices), len(col_indices)
        return new_obj

    def __copy__(self):
        return PandasOnMultiprocessFramePartition(
            self.pid, self._length_cache, self._width_cache
        )

    def to_pandas(self):
        """Convert the object stored in this partition to a Pandas DataFrame.

        Note: If the underlying object is a Pandas DataFrame, this will likely
            only need to call `get`

        Returns:
            A Pandas DataFrame.
        """
        dataframe = self.get()
        assert type(dataframe) is pandas.DataFrame or type(dataframe) is pandas.Series

        return dataframe

    @classmethod
    def put(cls, obj):
        """A factory classmethod to format a given object.

        Args:
            obj: An object.

        Returns:
            A `RemotePartitions` object.
        """
        pid = put_mp(obj)
        return cls(pid)

    @classmethod
    def preprocess_func(cls, func):
        """Preprocess a function before an `apply` call.

        Note: This is a classmethod because the definition of how to preprocess
            should be class-wide. Also, we may want to use this before we
            deploy a preprocessed function to multiple `BaseFramePartition`
            objects.

        Args:
            func: The function to preprocess.

        Returns:
            An object that can be accepted by `apply`.
        """
        import dill
        return dill.dumps(func)

    @classmethod
    def length_extraction_fn(cls):
        """The function to compute the length of the object in this partition.

        Returns:
            A callable function.
        """
        return length_fn_pandas

    @classmethod
    def width_extraction_fn(cls):
        """The function to compute the width of the object in this partition.

        Returns:
            A callable function.
        """
        return width_fn_pandas

    _length_cache = None
    _width_cache = None

    def length(self):
        if self._length_cache is None:
            self._length_cache = type(self).length_extraction_fn()(self.get())
        return self._length_cache

    def width(self):
        if self._width_cache is None:
            self._width_cache = type(self).width_extraction_fn()(self.get())
        return self._width_cache

    @classmethod
    def empty(cls):
        return cls(pandas.DataFrame())
```
That looks great @eavidan, would cloudpickle work instead of dill? It is likely because we use lambda functions internally, and if I remember correctly, pickle doesn't support lambdas. 

Feel free to open a WIP PR, that way I can comment and pull the code to test/play around.

Commit message: Upgrade dask versions

Patch to review:
diff --git a/modin/data_management/factories.py b/modin/data_management/factories.py
index 0e874b0f506..329b60bf56a 100644
--- a/modin/data_management/factories.py
+++ b/modin/data_management/factories.py
@@ -208,7 +208,7 @@ class PandasOnPythonFactory(BaseFactory):
 
 class PandasOnDaskFactory(BaseFactory):
 
-    from modin.engines.dask.pandas_on_dask_delayed.io import PandasOnDaskIO
+    from modin.engines.dask.pandas_on_dask_futures.io import PandasOnDaskIO
 
     io_cls = PandasOnDaskIO
 
diff --git a/modin/engines/dask/pandas_on_dask_futures/__init__.py b/modin/engines/dask/pandas_on_dask_futures/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/modin/engines/dask/pandas_on_dask_futures/frame/__init__.py b/modin/engines/dask/pandas_on_dask_futures/frame/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/modin/engines/dask/pandas_on_dask_futures/frame/axis_partition.py b/modin/engines/dask/pandas_on_dask_futures/frame/axis_partition.py
new file mode 100644
index 00000000000..1a5bf506e0a
--- /dev/null
+++ b/modin/engines/dask/pandas_on_dask_futures/frame/axis_partition.py
@@ -0,0 +1,74 @@
+# from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+from distributed.client import _get_global_client
+from distributed import Future
+
+from modin.engines.base.frame.axis_partition import PandasFrameAxisPartition
+from .partition import PandasOnDaskFramePartition
+
+
+class PandasOnDaskFrameAxisPartition(PandasFrameAxisPartition):
+    def __init__(self, list_of_blocks):
+        # Unwrap from BaseFramePartition object for ease of use
+        for obj in list_of_blocks:
+            obj.drain_call_queue()
+        self.list_of_blocks = [obj.future for obj in list_of_blocks]
+
+    partition_type = PandasOnDaskFramePartition
+    instance_type = Future
+
+    @classmethod
+    def deploy_axis_func(
+        cls, axis, func, num_splits, kwargs, maintain_partitioning, *partitions
+    ):
+        client = _get_global_client()
+        axis_result = client.submit(
+            PandasFrameAxisPartition.deploy_axis_func,
+            axis,
+            func,
+            num_splits,
+            kwargs,
+            maintain_partitioning,
+            *partitions
+        )
+        # We have to do this to split it back up. It is already split, but we need to
+        # get futures for each.
+        return [client.submit(lambda l: l[i], axis_result) for i in range(num_splits)]
+
+    @classmethod
+    def deploy_func_between_two_axis_partitions(
+        cls, axis, func, num_splits, len_of_left, kwargs, *partitions
+    ):
+        client = _get_global_client()
+        axis_result = client.submit(
+            PandasFrameAxisPartition.deploy_func_between_two_axis_partitions,
+            axis,
+            func,
+            num_splits,
+            len_of_left,
+            kwargs,
+            *partitions
+        )
+        # We have to do this to split it back up. It is already split, but we need to
+        # get futures for each.
+        return [client.submit(lambda l: l[i], axis_result) for i in range(num_splits)]
+
+
+class PandasOnDaskFrameColumnPartition(PandasOnDaskFrameAxisPartition):
+    """The column partition implementation for Multiprocess. All of the implementation
+        for this class is in the parent class, and this class defines the axis
+        to perform the computation over.
+    """
+
+    axis = 0
+
+
+class PandasOnDaskFrameRowPartition(PandasOnDaskFrameAxisPartition):
+    """The row partition implementation for Multiprocess. All of the implementation
+        for this class is in the parent class, and this class defines the axis
+        to perform the computation over.
+    """
+
+    axis = 1
diff --git a/modin/engines/dask/pandas_on_dask_futures/frame/data.py b/modin/engines/dask/pandas_on_dask_futures/frame/data.py
new file mode 100644
index 00000000000..ae8a6cd2836
--- /dev/null
+++ b/modin/engines/dask/pandas_on_dask_futures/frame/data.py
@@ -0,0 +1,33 @@
+from distributed.client import _get_global_client
+
+from modin.engines.base.frame.data import BasePandasFrame
+from .partition_manager import DaskFrameManager
+
+
+class PandasOnDaskFrame(BasePandasFrame):
+
+    _frame_mgr_cls = DaskFrameManager
+
+    @property
+    def _row_lengths(self):
+        """Compute the row lengths if they are not cached.
+
+        Returns:
+            A list of row lengths.
+        """
+        client = _get_global_client()
+        if self._row_lengths_cache is None:
+            self._row_lengths_cache = client.gather([obj.apply(lambda df: len(df)).future for obj in self._partitions.T[0]])
+        return self._row_lengths_cache
+
+    @property
+    def _column_widths(self):
+        """Compute the column widths if they are not cached.
+
+        Returns:
+            A list of column widths.
+        """
+        client = _get_global_client()
+        if self._column_widths_cache is None:
+            self._column_widths_cache = client.gather([obj.apply(lambda df: len(df.columns)).future for obj in self._partitions[0]])
+        return self._column_widths_cache
diff --git a/modin/engines/dask/pandas_on_dask_futures/frame/partition.py b/modin/engines/dask/pandas_on_dask_futures/frame/partition.py
new file mode 100644
index 00000000000..55fe121f4b6
--- /dev/null
+++ b/modin/engines/dask/pandas_on_dask_futures/frame/partition.py
@@ -0,0 +1,187 @@
+import pandas
+from distributed.client import _get_global_client
+
+from modin.engines.base.frame.partition import BaseFramePartition
+from modin.data_management.utils import length_fn_pandas, width_fn_pandas
+
+
+def apply_list_of_funcs(funcs, df):
+    for func, kwargs in funcs:
+        df = func(df, **kwargs)
+    return df
+
+
+class PandasOnDaskFramePartition(BaseFramePartition):
+    """This abstract class holds the data and metadata for a single partition.
+        The methods required for implementing this abstract class are listed in
+        the section immediately following this.
+
+        The API exposed by the children of this object is used in
+        `BaseFrameManager`.
+
+        Note: These objects are treated as immutable by `BaseFrameManager`
+        subclasses. There is no logic for updating inplace.
+    """
+
+    def __init__(self, future, length=None, width=None, call_queue=None):
+        self.future = future
+        if call_queue is None:
+            call_queue = []
+        self.call_queue = call_queue
+        self._length_cache = length
+        self._width_cache = width
+        self.client = _get_global_client()
+
+    def get(self):
+        """Flushes the call_queue and returns the data.
+
+        Note: Since this object is a simple wrapper, just return the data.
+
+        Returns:
+            The object that was `put`.
+        """
+        self.drain_call_queue()
+        # blocking operation
+        if isinstance(self.future, pandas.DataFrame):
+            return self.future
+        return self.future.result()
+
+    def apply(self, func, **kwargs):
+        """Apply some callable function to the data in this partition.
+
+        Note: It is up to the implementation how kwargs are handled. They are
+            an important part of many implementations. As of right now, they
+            are not serialized.
+
+        Args:
+            func: The lambda to apply (may already be correctly formatted)
+
+        Returns:
+             A new `BaseFramePartition` containing the object that has had `func`
+             applied to it.
+        """
+        call_queue = self.call_queue + [[func, kwargs]]
+        future = self.client.submit(apply_list_of_funcs, call_queue, self.future)
+        return PandasOnDaskFramePartition(future)
+
+    def add_to_apply_calls(self, func, **kwargs):
+        return PandasOnDaskFramePartition(
+            self.future, call_queue=self.call_queue + [[func, kwargs]]
+        )
+
+    def drain_call_queue(self):
+        if len(self.call_queue) == 0:
+            return
+        self.future = self.apply(lambda x: x).future
+
+    def mask(self, row_indices, col_indices):
+        new_obj = self.add_to_apply_calls(
+            lambda df: pandas.DataFrame(df.iloc[row_indices, col_indices])
+        )
+        new_obj._length_cache = (
+            len(row_indices)
+            if not isinstance(row_indices, slice)
+            else self._length_cache
+        )
+        new_obj._width_cache = (
+            len(col_indices)
+            if not isinstance(col_indices, slice)
+            else self._width_cache
+        )
+        return new_obj
+
+    def __copy__(self):
+        return PandasOnDaskFramePartition(
+            self.future, self._length_cache, self._width_cache
+        )
+
+    def to_pandas(self):
+        """Convert the object stored in this partition to a Pandas DataFrame.
+
+        Note: If the underlying object is a Pandas DataFrame, this will likely
+            only need to call `get`
+
+        Returns:
+            A Pandas DataFrame.
+        """
+        dataframe = self.get()
+        assert type(dataframe) is pandas.DataFrame or type(dataframe) is pandas.Series
+
+        return dataframe
+
+    def to_numpy(self):
+        """Convert the object stored in this parition to a Numpy Array.
+
+        Returns:
+            A Numpy Array.
+        """
+        return self.apply(lambda df: df.to_numpy()).get()
+
+    @classmethod
+    def put(cls, obj):
+        """A factory classmethod to format a given object.
+
+        Args:
+            obj: An object.
+
+        Returns:
+            A `RemotePartitions` object.
+        """
+        client = _get_global_client()
+        return cls(client.scatter(obj))
+
+    @classmethod
+    def preprocess_func(cls, func):
+        """Preprocess a function before an `apply` call.
+
+        Note: This is a classmethod because the definition of how to preprocess
+            should be class-wide. Also, we may want to use this before we
+            deploy a preprocessed function to multiple `BaseFramePartition`
+            objects.
+
+        Args:
+            func: The function to preprocess.
+
+        Returns:
+            An object that can be accepted by `apply`.
+        """
+        return func
+
+    @classmethod
+    def length_extraction_fn(cls):
+        """The function to compute the length of the object in this partition.
+
+        Returns:
+            A callable function.
+        """
+        return length_fn_pandas
+
+    @classmethod
+    def width_extraction_fn(cls):
+        """The function to compute the width of the object in this partition.
+
+        Returns:
+            A callable function.
+        """
+        return width_fn_pandas
+
+    _length_cache = None
+    _width_cache = None
+
+    def length(self):
+        if self._length_cache is None:
+            self._length_cache = self.apply(lambda df: len(df)).future
+        if isinstance(self._length_cache, type(self.future)):
+            self._length_cache = self._length_cache.result()
+        return self._length_cache
+
+    def width(self):
+        if self._width_cache is None:
+            self._width_cache = self.apply(lambda df: len(df.columns)).future
+        if isinstance(self._width_cache, type(self.future)):
+            self._width_cache = self._width_cache.result()
+        return self._width_cache
+
+    @classmethod
+    def empty(cls):
+        return cls(pandas.DataFrame(), 0, 0)
diff --git a/modin/engines/dask/pandas_on_dask_futures/frame/partition_manager.py b/modin/engines/dask/pandas_on_dask_futures/frame/partition_manager.py
new file mode 100644
index 00000000000..4b1e8aa4fde
--- /dev/null
+++ b/modin/engines/dask/pandas_on_dask_futures/frame/partition_manager.py
@@ -0,0 +1,55 @@
+from distributed import get_client
+
+
+from modin.engines.base.frame.partition_manager import BaseFrameManager
+from .axis_partition import (
+    PandasOnDaskFrameColumnPartition,
+    PandasOnDaskFrameRowPartition,
+)
+from .partition import PandasOnDaskFramePartition
+from modin.error_message import ErrorMessage
+
+
+class DaskFrameManager(BaseFrameManager):
+    """This class implements the interface in `BaseFrameManager`."""
+
+    # This object uses RayRemotePartition objects as the underlying store.
+    _partition_class = PandasOnDaskFramePartition
+    _column_partitions_class = PandasOnDaskFrameColumnPartition
+    _row_partition_class = PandasOnDaskFrameRowPartition
+
+    @classmethod
+    def get_indices(cls, axis, partitions, index_func=None):
+        """This gets the internal indices stored in the partitions.
+
+        Note: These are the global indices of the object. This is mostly useful
+            when you have deleted rows/columns internally, but do not know
+            which ones were deleted.
+
+        Args:
+            axis: This axis to extract the labels. (0 - index, 1 - columns).
+            index_func: The function to be used to extract the function.
+            old_blocks: An optional previous object that this object was
+                created from. This is used to compute the correct offsets.
+
+        Returns:
+            A Pandas Index object.
+        """
+        client = get_client()
+        ErrorMessage.catch_bugs_and_request_email(not callable(index_func))
+        func = cls.preprocess_func(index_func)
+        if axis == 0:
+            # We grab the first column of blocks and extract the indices
+            new_idx = (
+                [idx.apply(func).future for idx in partitions.T[0]]
+                if len(partitions.T)
+                else []
+            )
+        else:
+            new_idx = (
+                [idx.apply(func).future for idx in partitions[0]]
+                if len(partitions)
+                else []
+            )
+        new_idx = client.gather(new_idx)
+        return new_idx[0].append(new_idx[1:]) if len(new_idx) else new_idx
diff --git a/modin/engines/dask/pandas_on_dask_futures/io.py b/modin/engines/dask/pandas_on_dask_futures/io.py
new file mode 100644
index 00000000000..b3689ec85aa
--- /dev/null
+++ b/modin/engines/dask/pandas_on_dask_futures/io.py
@@ -0,0 +1,13 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+from modin.engines.base.io import BaseIO
+from modin.backends.pandas.query_compiler import PandasQueryCompiler
+from modin.engines.dask.pandas_on_dask_futures.frame.data import PandasOnDaskFrame
+
+
+class PandasOnDaskIO(BaseIO):
+
+    frame_cls = PandasOnDaskFrame
+    query_compiler_cls = PandasQueryCompiler
diff --git a/modin/engines/dask/pandas_on_dask_futures/series/__init__.py b/modin/engines/dask/pandas_on_dask_futures/series/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/modin/engines/python/pandas_on_python/frame/partition.py b/modin/engines/python/pandas_on_python/frame/partition.py
index 6ba5131c116..70d66ab2b72 100644
--- a/modin/engines/python/pandas_on_python/frame/partition.py
+++ b/modin/engines/python/pandas_on_python/frame/partition.py
@@ -16,8 +16,10 @@ class PandasOnPythonFramePartition(BaseFramePartition):
         subclasses. There is no logic for updating inplace.
     """
 
-    def __init__(self, data, length=None, width=None, call_queue=[]):
+    def __init__(self, data, length=None, width=None, call_queue=None):
         self.data = data
+        if call_queue is None:
+            call_queue = []
         self.call_queue = call_queue
         self._length_cache = length
         self._width_cache = width
diff --git a/modin/pandas/test/test_dataframe.py b/modin/pandas/test/test_dataframe.py
index a69ec26429e..fd2c833665b 100644
--- a/modin/pandas/test/test_dataframe.py
+++ b/modin/pandas/test/test_dataframe.py
@@ -332,6 +332,8 @@ def comparison_inter_ops_helper(self, modin_df, pandas_df, op):
                 getattr(modin_df, op)(modin_df)
         else:
             modin_result = getattr(modin_df, op)(modin_df)
+            print(modin_result)
+            print(pandas_result)
             df_equals(modin_result, pandas_result)
 
         try:
diff --git a/requirements.txt b/requirements.txt
index eaa061077b8..fb5723129eb 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,7 +1,7 @@
 pandas==0.25.1
 numpy
-dask[complete]==1.1.0
-distributed==1.25.0
+dask[complete]>=2.1.0
+distributed>=2.3.2
 ray==0.7.3
 psutil==5.4.8
 strip_hints==0.1.1
diff --git a/setup.py b/setup.py
index 604eae54ec7..69600c9e115 100644
--- a/setup.py
+++ b/setup.py
@@ -18,6 +18,6 @@
     install_requires=["pandas==0.25.1", "ray==0.7.3"],
     extras_require={
         # can be installed by pip install modin[dask]
-        "dask": ["dask==1.1.0", "distributed==1.25.0"]
+        "dask": ["dask>=2.1.0", "distributed>=2.3.2"]
     },
 )


Code review comments:
diff_hunk: @@ -332,6 +332,8 @@ def comparison_inter_ops_helper(self, modin_df, pandas_df, op):
                 getattr(modin_df, op)(modin_df)
         else:
             modin_result = getattr(modin_df, op)(modin_df)
+            print(modin_result)
path: modin/pandas/test/test_dataframe.py
text: @user1:
extraneous print

diff_hunk: @@ -0,0 +1,55 @@
+from distributed import get_client
+
+
+from modin.engines.base.frame.partition_manager import BaseFrameManager
+from .axis_partition import (
+    PandasOnDaskFrameColumnPartition,
+    PandasOnDaskFrameRowPartition,
+)
+from .partition import PandasOnDaskFramePartition
+from modin.error_message import ErrorMessage
+
+
+class DaskFrameManager(BaseFrameManager):
+    """This class implements the interface in `BaseFrameManager`."""
+
+    # This object uses RayRemotePartition objects as the underlying store.
+    _partition_class = PandasOnDaskFramePartition
+    _column_partitions_class = PandasOnDaskFrameColumnPartition
+    _row_partition_class = PandasOnDaskFrameRowPartition
+
+    @classmethod
+    def get_indices(cls, axis, partitions, index_func=None):
path: modin/engines/dask/pandas_on_dask_futures/frame/partition_manager.py
text: @user1:
Since `index_func` has to be a callable,
```suggestion
    def get_indices(cls, axis, partitions, index_func):
```