Update ray and make changes to be compatible with API changes (#284)

modin/data_management/partitioning/axis_partition/pandas_on_ray.py

@@ -35,7 +35,7 @@ def apply(self, func, num_splits=None, other_axis_partition=None, **kwargs):
         if other_axis_partition is not None:
             return [
                 PandasOnRayRemotePartition(obj)
-                for obj in deploy_ray_func_between_two_axis_partitions._submit(
+                for obj in deploy_ray_func_between_two_axis_partitions._remote(
                     args=(self.axis, func, num_splits, len(self.list_of_blocks), kwargs)
                     + tuple(self.list_of_blocks + other_axis_partition.list_of_blocks),
                     num_return_vals=num_splits,
@@ -46,7 +46,7 @@ def apply(self, func, num_splits=None, other_axis_partition=None, **kwargs):
         args.extend(self.list_of_blocks)
         return [
             PandasOnRayRemotePartition(obj)
-            for obj in deploy_ray_axis_func._submit(args, num_return_vals=num_splits)
+            for obj in deploy_ray_axis_func._remote(args, num_return_vals=num_splits)
         ]
 
     def shuffle(self, func, num_splits=None, **kwargs):
@@ -66,7 +66,7 @@ def shuffle(self, func, num_splits=None, **kwargs):
         args.extend(self.list_of_blocks)
         return [
             PandasOnRayRemotePartition(obj)
-            for obj in deploy_ray_axis_func._submit(args, num_return_vals=num_splits)
+            for obj in deploy_ray_axis_func._remote(args, num_return_vals=num_splits)
         ]
 
 

modin/pandas/__init__.py

@@ -81,7 +81,6 @@
                 redirect_output=True,
                 include_webui=False,
                 redirect_worker_output=True,
-                use_raylet=True,
                 ignore_reinit_error=True,
             )
     except AssertionError:

modin/pandas/io.py

@@ -69,7 +69,7 @@ def _read_parquet_pandas_on_ray(path, engine, columns, **kwargs):
     # We need to transpose the oids array to fit our schema.
     blk_partitions = np.array(
         [
-            _read_parquet_columns._submit(
+            _read_parquet_columns._remote(
                 args=(path, cols, num_splits, kwargs), num_return_vals=num_splits + 1
             )
             for cols in col_partitions
@@ -178,7 +178,7 @@ def _read_csv_from_file_pandas_on_ray(filepath, kwargs={}):
             start = f.tell()
             f.seek(chunk_size, os.SEEK_CUR)
             f.readline()  # Read a whole number of lines
-            partition_id = _read_csv_with_offset_pandas_on_ray._submit(
+            partition_id = _read_csv_with_offset_pandas_on_ray._remote(
                 args=(
                     filepath,
                     num_splits,
@@ -552,7 +552,7 @@ def read_hdf(path_or_buf, key=None, mode="r", columns=None):
     ]
     blk_partitions = np.array(
         [
-            _read_hdf_columns._submit(
+            _read_hdf_columns._remote(
                 args=(path_or_buf, cols, num_splits, key, mode),
                 num_return_vals=num_splits + 1,
             )

requirements.txt

@@ -1,5 +1,4 @@
 pandas==0.23.4
-redis==2.10.6
-ray==0.5.3
+ray==0.6.0
 strip_hints
 typing

setup.py

@@ -10,10 +10,9 @@
 setup(
     name="modin",
     version="0.2.4",
-    description="Modin: Pandas on Ray - Make your pandas code run faster with "
-                "a single line of code change.",
+    description="Modin: Make your pandas code run faster by changing one line of code.",
     packages=find_packages(),
     url="https://github.com/modin-project/modin",
     long_description=long_description,
     long_description_content_type="text/markdown",
-    install_requires=["pandas==0.23.4", "redis==2.10.6", "ray==0.5.3"])
+    install_requires=["pandas==0.23.4", "ray==0.6.0"])