We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Modin not working on latest ray master
With

```python
import ray
import modin.pandas as pd
frame_data = [1, 2, 3, 4, 5, 6, 7, 8]
frame = pd.DataFrame(frame_data)
assert frame.sum().squeeze() == sum(frame_data)
```

we get the following error:

```
Traceback (most recent call last):
  File "python/ray/test/test_modin.py", line 16, in <module>
    test_modin_import()
  File "python/ray/test/test_modin.py", line 9, in test_modin_import
    import modin.pandas as pd
  File "/home/travis/.local/lib/python3.6/site-packages/ray-0.7.0.dev0-py3.6-linux-x86_64.egg/ray/modin/modin/pandas/__init__.py", line 134, in <module>
    initialize_ray()
  File "/home/travis/.local/lib/python3.6/site-packages/ray-0.7.0.dev0-py3.6-linux-x86_64.egg/ray/modin/modin/pandas/__init__.py", line 126, in initialize_ray
    object_store_memory=object_store_memory,
  File "/home/travis/.local/lib/python3.6/site-packages/ray-0.7.0.dev0-py3.6-linux-x86_64.egg/ray/worker.py", line 1427, in init
    _internal_config=_internal_config,
  File "/home/travis/.local/lib/python3.6/site-packages/ray-0.7.0.dev0-py3.6-linux-x86_64.egg/ray/parameter.py", line 145, in __init__
    self._check_usage()
  File "/home/travis/.local/lib/python3.6/site-packages/ray-0.7.0.dev0-py3.6-linux-x86_64.egg/ray/parameter.py", line 189, in _check_usage
    "The redirect_worker_output argument is deprecated. To "
DeprecationWarning: The redirect_worker_output argument is deprecated. To control logging to the driver, use the 'log_to_driver' argument to 'ray.init()'
```

see also https://github.com/ray-project/ray/pull/4069 and https://github.com/ray-project/ray/commit/5f717518914d303126ed1bac3752150c0284fc3e.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `modin/pandas/__init__.py`
Content:
```
1 from __future__ import absolute_import
2 from __future__ import division
3 from __future__ import print_function
4 
5 # TODO: In the future `set_option` or similar needs to run on every node
6 # in order to keep all pandas instances across nodes consistent
7 import pandas
8 from pandas import (
9     eval,
10     unique,
11     value_counts,
12     cut,
13     to_numeric,
14     factorize,
15     test,
16     qcut,
17     match,
18     Panel,
19     date_range,
20     period_range,
21     Index,
22     MultiIndex,
23     CategoricalIndex,
24     Series,
25     bdate_range,
26     DatetimeIndex,
27     Timedelta,
28     Timestamp,
29     to_timedelta,
30     set_eng_float_format,
31     set_option,
32     NaT,
33     PeriodIndex,
34     Categorical,
35 )
36 import threading
37 import os
38 import ray
39 import types
40 
41 from .. import __version__
42 from .concat import concat
43 from .dataframe import DataFrame
44 from .datetimes import to_datetime
45 from .io import (
46     read_csv,
47     read_parquet,
48     read_json,
49     read_html,
50     read_clipboard,
51     read_excel,
52     read_hdf,
53     read_feather,
54     read_msgpack,
55     read_stata,
56     read_sas,
57     read_pickle,
58     read_sql,
59     read_gbq,
60     read_table,
61 )
62 from .reshape import get_dummies, melt, crosstab
63 from .general import isna, isnull, merge, pivot_table
64 from .plotting import Plotting as plotting
65 from .. import __execution_engine__ as execution_engine
66 
67 __pandas_version__ = "0.23.4"
68 
69 if pandas.__version__ != __pandas_version__:
70     raise ImportError(
71         "The pandas version installed does not match the required pandas "
72         "version in Modin. Please install pandas {} to use "
73         "Modin.".format(__pandas_version__)
74     )
75 
76 # Set this so that Pandas doesn't try to multithread by itself
77 os.environ["OMP_NUM_THREADS"] = "1"
78 num_cpus = 1
79 
80 
81 def initialize_ray():
82     """Initializes ray based on environment variables and internal defaults."""
83     if threading.current_thread().name == "MainThread":
84         plasma_directory = None
85         object_store_memory = None
86         if "MODIN_MEMORY" in os.environ:
87             object_store_memory = os.environ["MODIN_MEMORY"]
88         if (
89             "MODIN_OUT_OF_CORE" in os.environ
90             and os.environ["MODIN_OUT_OF_CORE"].title() == "True"
91         ):
92             from tempfile import gettempdir
93 
94             plasma_directory = gettempdir()
95             # We may have already set the memory from the environment variable, we don't
96             # want to overwrite that value if we have.
97             if object_store_memory is None:
98                 try:
99                     from psutil import virtual_memory
100                 except ImportError:
101                     raise ImportError(
102                         "To use Modin out of core, please install modin[out_of_core]: "
103                         '`pip install "modin[out_of_core]"`'
104                     )
105                 # Round down to the nearest Gigabyte.
106                 mem_bytes = virtual_memory().total // 10 ** 9 * 10 ** 9
107                 # Default to 8x memory for out of core
108                 object_store_memory = 8 * mem_bytes
109         elif "MODIN_MEMORY" in os.environ:
110             object_store_memory = os.environ["MODIN_MEMORY"]
111         # In case anything failed above, we can still improve the memory for Modin.
112         if object_store_memory is None:
113             # Round down to the nearest Gigabyte.
114             object_store_memory = int(
115                 0.6 * ray.utils.get_system_memory() // 10 ** 9 * 10 ** 9
116             )
117             # If the memory pool is smaller than 2GB, just use the default in ray.
118             if object_store_memory == 0:
119                 object_store_memory = None
120         ray.init(
121             redirect_output=True,
122             include_webui=False,
123             redirect_worker_output=True,
124             ignore_reinit_error=True,
125             plasma_directory=plasma_directory,
126             object_store_memory=object_store_memory,
127         )
128         # Register custom serializer for method objects to avoid warning message.
129         # We serialize `MethodType` objects when we use AxisPartition operations.
130         ray.register_custom_serializer(types.MethodType, use_pickle=True)
131 
132 
133 if execution_engine == "Ray":
134     initialize_ray()
135     num_cpus = ray.global_state.cluster_resources()["CPU"]
136 elif execution_engine == "Dask":
137     from distributed.client import _get_global_client
138 
139     if threading.current_thread().name == "MainThread":
140         # initialize the dask client
141         client = _get_global_client()
142         if client is None:
143             from distributed import Client
144 
145             client = Client()
146         num_cpus = sum(client.ncores().values())
147 elif execution_engine != "Python":
148     raise ImportError("Unrecognized execution engine: {}.".format(execution_engine))
149 
150 DEFAULT_NPARTITIONS = max(4, int(num_cpus))
151 
152 __all__ = [
153     "DataFrame",
154     "Series",
155     "read_csv",
156     "read_parquet",
157     "read_json",
158     "read_html",
159     "read_clipboard",
160     "read_excel",
161     "read_hdf",
162     "read_feather",
163     "read_msgpack",
164     "read_stata",
165     "read_sas",
166     "read_pickle",
167     "read_sql",
168     "read_gbq",
169     "read_table",
170     "concat",
171     "eval",
172     "unique",
173     "value_counts",
174     "cut",
175     "to_numeric",
176     "factorize",
177     "test",
178     "qcut",
179     "match",
180     "to_datetime",
181     "get_dummies",
182     "isna",
183     "isnull",
184     "merge",
185     "pivot_table",
186     "Panel",
187     "date_range",
188     "Index",
189     "MultiIndex",
190     "Series",
191     "bdate_range",
192     "period_range",
193     "DatetimeIndex",
194     "to_timedelta",
195     "set_eng_float_format",
196     "set_option",
197     "CategoricalIndex",
198     "Timedelta",
199     "Timestamp",
200     "NaT",
201     "PeriodIndex",
202     "Categorical",
203     "__version__",
204     "melt",
205     "crosstab",
206     "plotting",
207 ]
208 
209 del pandas
210
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/modin/pandas/__init__.py b/modin/pandas/__init__.py
--- a/modin/pandas/__init__.py
+++ b/modin/pandas/__init__.py
@@ -118,9 +118,7 @@
             if object_store_memory == 0:
                 object_store_memory = None
         ray.init(
-            redirect_output=True,
             include_webui=False,
-            redirect_worker_output=True,
             ignore_reinit_error=True,
             plasma_directory=plasma_directory,
             object_store_memory=object_store_memory,

{"golden_diff": "diff --git a/modin/pandas/__init__.py b/modin/pandas/__init__.py\n--- a/modin/pandas/__init__.py\n+++ b/modin/pandas/__init__.py\n@@ -118,9 +118,7 @@\n             if object_store_memory == 0:\n                 object_store_memory = None\n         ray.init(\n-            redirect_output=True,\n             include_webui=False,\n-            redirect_worker_output=True,\n             ignore_reinit_error=True,\n             plasma_directory=plasma_directory,\n             object_store_memory=object_store_memory,\n", "issue": "Modin not working on latest ray master\nWith\r\n\r\n```python\r\nimport ray\r\nimport modin.pandas as pd\r\nframe_data = [1, 2, 3, 4, 5, 6, 7, 8]\r\nframe = pd.DataFrame(frame_data)\r\nassert frame.sum().squeeze() == sum(frame_data)\r\n```\r\n\r\nwe get the following error:\r\n\r\n```\r\nTraceback (most recent call last):\r\n  File \"python/ray/test/test_modin.py\", line 16, in <module>\r\n    test_modin_import()\r\n  File \"python/ray/test/test_modin.py\", line 9, in test_modin_import\r\n    import modin.pandas as pd\r\n  File \"/home/travis/.local/lib/python3.6/site-packages/ray-0.7.0.dev0-py3.6-linux-x86_64.egg/ray/modin/modin/pandas/__init__.py\", line 134, in <module>\r\n    initialize_ray()\r\n  File \"/home/travis/.local/lib/python3.6/site-packages/ray-0.7.0.dev0-py3.6-linux-x86_64.egg/ray/modin/modin/pandas/__init__.py\", line 126, in initialize_ray\r\n    object_store_memory=object_store_memory,\r\n  File \"/home/travis/.local/lib/python3.6/site-packages/ray-0.7.0.dev0-py3.6-linux-x86_64.egg/ray/worker.py\", line 1427, in init\r\n    _internal_config=_internal_config,\r\n  File \"/home/travis/.local/lib/python3.6/site-packages/ray-0.7.0.dev0-py3.6-linux-x86_64.egg/ray/parameter.py\", line 145, in __init__\r\n    self._check_usage()\r\n  File \"/home/travis/.local/lib/python3.6/site-packages/ray-0.7.0.dev0-py3.6-linux-x86_64.egg/ray/parameter.py\", line 189, in _check_usage\r\n    \"The redirect_worker_output argument is deprecated. To \"\r\nDeprecationWarning: The redirect_worker_output argument is deprecated. To control logging to the driver, use the 'log_to_driver' argument to 'ray.init()'\r\n```\r\n\r\nsee also https://github.com/ray-project/ray/pull/4069 and https://github.com/ray-project/ray/commit/5f717518914d303126ed1bac3752150c0284fc3e.\n", "before_files": [{"content": "from __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\n# TODO: In the future `set_option` or similar needs to run on every node\n# in order to keep all pandas instances across nodes consistent\nimport pandas\nfrom pandas import (\n    eval,\n    unique,\n    value_counts,\n    cut,\n    to_numeric,\n    factorize,\n    test,\n    qcut,\n    match,\n    Panel,\n    date_range,\n    period_range,\n    Index,\n    MultiIndex,\n    CategoricalIndex,\n    Series,\n    bdate_range,\n    DatetimeIndex,\n    Timedelta,\n    Timestamp,\n    to_timedelta,\n    set_eng_float_format,\n    set_option,\n    NaT,\n    PeriodIndex,\n    Categorical,\n)\nimport threading\nimport os\nimport ray\nimport types\n\nfrom .. import __version__\nfrom .concat import concat\nfrom .dataframe import DataFrame\nfrom .datetimes import to_datetime\nfrom .io import (\n    read_csv,\n    read_parquet,\n    read_json,\n    read_html,\n    read_clipboard,\n    read_excel,\n    read_hdf,\n    read_feather,\n    read_msgpack,\n    read_stata,\n    read_sas,\n    read_pickle,\n    read_sql,\n    read_gbq,\n    read_table,\n)\nfrom .reshape import get_dummies, melt, crosstab\nfrom .general import isna, isnull, merge, pivot_table\nfrom .plotting import Plotting as plotting\nfrom .. import __execution_engine__ as execution_engine\n\n__pandas_version__ = \"0.23.4\"\n\nif pandas.__version__ != __pandas_version__:\n    raise ImportError(\n        \"The pandas version installed does not match the required pandas \"\n        \"version in Modin. Please install pandas {} to use \"\n        \"Modin.\".format(__pandas_version__)\n    )\n\n# Set this so that Pandas doesn't try to multithread by itself\nos.environ[\"OMP_NUM_THREADS\"] = \"1\"\nnum_cpus = 1\n\n\ndef initialize_ray():\n    \"\"\"Initializes ray based on environment variables and internal defaults.\"\"\"\n    if threading.current_thread().name == \"MainThread\":\n        plasma_directory = None\n        object_store_memory = None\n        if \"MODIN_MEMORY\" in os.environ:\n            object_store_memory = os.environ[\"MODIN_MEMORY\"]\n        if (\n            \"MODIN_OUT_OF_CORE\" in os.environ\n            and os.environ[\"MODIN_OUT_OF_CORE\"].title() == \"True\"\n        ):\n            from tempfile import gettempdir\n\n            plasma_directory = gettempdir()\n            # We may have already set the memory from the environment variable, we don't\n            # want to overwrite that value if we have.\n            if object_store_memory is None:\n                try:\n                    from psutil import virtual_memory\n                except ImportError:\n                    raise ImportError(\n                        \"To use Modin out of core, please install modin[out_of_core]: \"\n                        '`pip install \"modin[out_of_core]\"`'\n                    )\n                # Round down to the nearest Gigabyte.\n                mem_bytes = virtual_memory().total // 10 ** 9 * 10 ** 9\n                # Default to 8x memory for out of core\n                object_store_memory = 8 * mem_bytes\n        elif \"MODIN_MEMORY\" in os.environ:\n            object_store_memory = os.environ[\"MODIN_MEMORY\"]\n        # In case anything failed above, we can still improve the memory for Modin.\n        if object_store_memory is None:\n            # Round down to the nearest Gigabyte.\n            object_store_memory = int(\n                0.6 * ray.utils.get_system_memory() // 10 ** 9 * 10 ** 9\n            )\n            # If the memory pool is smaller than 2GB, just use the default in ray.\n            if object_store_memory == 0:\n                object_store_memory = None\n        ray.init(\n            redirect_output=True,\n            include_webui=False,\n            redirect_worker_output=True,\n            ignore_reinit_error=True,\n            plasma_directory=plasma_directory,\n            object_store_memory=object_store_memory,\n        )\n        # Register custom serializer for method objects to avoid warning message.\n        # We serialize `MethodType` objects when we use AxisPartition operations.\n        ray.register_custom_serializer(types.MethodType, use_pickle=True)\n\n\nif execution_engine == \"Ray\":\n    initialize_ray()\n    num_cpus = ray.global_state.cluster_resources()[\"CPU\"]\nelif execution_engine == \"Dask\":\n    from distributed.client import _get_global_client\n\n    if threading.current_thread().name == \"MainThread\":\n        # initialize the dask client\n        client = _get_global_client()\n        if client is None:\n            from distributed import Client\n\n            client = Client()\n        num_cpus = sum(client.ncores().values())\nelif execution_engine != \"Python\":\n    raise ImportError(\"Unrecognized execution engine: {}.\".format(execution_engine))\n\nDEFAULT_NPARTITIONS = max(4, int(num_cpus))\n\n__all__ = [\n    \"DataFrame\",\n    \"Series\",\n    \"read_csv\",\n    \"read_parquet\",\n    \"read_json\",\n    \"read_html\",\n    \"read_clipboard\",\n    \"read_excel\",\n    \"read_hdf\",\n    \"read_feather\",\n    \"read_msgpack\",\n    \"read_stata\",\n    \"read_sas\",\n    \"read_pickle\",\n    \"read_sql\",\n    \"read_gbq\",\n    \"read_table\",\n    \"concat\",\n    \"eval\",\n    \"unique\",\n    \"value_counts\",\n    \"cut\",\n    \"to_numeric\",\n    \"factorize\",\n    \"test\",\n    \"qcut\",\n    \"match\",\n    \"to_datetime\",\n    \"get_dummies\",\n    \"isna\",\n    \"isnull\",\n    \"merge\",\n    \"pivot_table\",\n    \"Panel\",\n    \"date_range\",\n    \"Index\",\n    \"MultiIndex\",\n    \"Series\",\n    \"bdate_range\",\n    \"period_range\",\n    \"DatetimeIndex\",\n    \"to_timedelta\",\n    \"set_eng_float_format\",\n    \"set_option\",\n    \"CategoricalIndex\",\n    \"Timedelta\",\n    \"Timestamp\",\n    \"NaT\",\n    \"PeriodIndex\",\n    \"Categorical\",\n    \"__version__\",\n    \"melt\",\n    \"crosstab\",\n    \"plotting\",\n]\n\ndel pandas\n", "path": "modin/pandas/__init__.py"}], "after_files": [{"content": "from __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\n# TODO: In the future `set_option` or similar needs to run on every node\n# in order to keep all pandas instances across nodes consistent\nimport pandas\nfrom pandas import (\n    eval,\n    unique,\n    value_counts,\n    cut,\n    to_numeric,\n    factorize,\n    test,\n    qcut,\n    match,\n    Panel,\n    date_range,\n    period_range,\n    Index,\n    MultiIndex,\n    CategoricalIndex,\n    Series,\n    bdate_range,\n    DatetimeIndex,\n    Timedelta,\n    Timestamp,\n    to_timedelta,\n    set_eng_float_format,\n    set_option,\n    NaT,\n    PeriodIndex,\n    Categorical,\n)\nimport threading\nimport os\nimport ray\nimport types\n\nfrom .. import __version__\nfrom .concat import concat\nfrom .dataframe import DataFrame\nfrom .datetimes import to_datetime\nfrom .io import (\n    read_csv,\n    read_parquet,\n    read_json,\n    read_html,\n    read_clipboard,\n    read_excel,\n    read_hdf,\n    read_feather,\n    read_msgpack,\n    read_stata,\n    read_sas,\n    read_pickle,\n    read_sql,\n    read_gbq,\n    read_table,\n)\nfrom .reshape import get_dummies, melt, crosstab\nfrom .general import isna, isnull, merge, pivot_table\nfrom .plotting import Plotting as plotting\nfrom .. import __execution_engine__ as execution_engine\n\n__pandas_version__ = \"0.23.4\"\n\nif pandas.__version__ != __pandas_version__:\n    raise ImportError(\n        \"The pandas version installed does not match the required pandas \"\n        \"version in Modin. Please install pandas {} to use \"\n        \"Modin.\".format(__pandas_version__)\n    )\n\n# Set this so that Pandas doesn't try to multithread by itself\nos.environ[\"OMP_NUM_THREADS\"] = \"1\"\nnum_cpus = 1\n\n\ndef initialize_ray():\n    \"\"\"Initializes ray based on environment variables and internal defaults.\"\"\"\n    if threading.current_thread().name == \"MainThread\":\n        plasma_directory = None\n        object_store_memory = None\n        if \"MODIN_MEMORY\" in os.environ:\n            object_store_memory = os.environ[\"MODIN_MEMORY\"]\n        if (\n            \"MODIN_OUT_OF_CORE\" in os.environ\n            and os.environ[\"MODIN_OUT_OF_CORE\"].title() == \"True\"\n        ):\n            from tempfile import gettempdir\n\n            plasma_directory = gettempdir()\n            # We may have already set the memory from the environment variable, we don't\n            # want to overwrite that value if we have.\n            if object_store_memory is None:\n                try:\n                    from psutil import virtual_memory\n                except ImportError:\n                    raise ImportError(\n                        \"To use Modin out of core, please install modin[out_of_core]: \"\n                        '`pip install \"modin[out_of_core]\"`'\n                    )\n                # Round down to the nearest Gigabyte.\n                mem_bytes = virtual_memory().total // 10 ** 9 * 10 ** 9\n                # Default to 8x memory for out of core\n                object_store_memory = 8 * mem_bytes\n        elif \"MODIN_MEMORY\" in os.environ:\n            object_store_memory = os.environ[\"MODIN_MEMORY\"]\n        # In case anything failed above, we can still improve the memory for Modin.\n        if object_store_memory is None:\n            # Round down to the nearest Gigabyte.\n            object_store_memory = int(\n                0.6 * ray.utils.get_system_memory() // 10 ** 9 * 10 ** 9\n            )\n            # If the memory pool is smaller than 2GB, just use the default in ray.\n            if object_store_memory == 0:\n                object_store_memory = None\n        ray.init(\n            include_webui=False,\n            ignore_reinit_error=True,\n            plasma_directory=plasma_directory,\n            object_store_memory=object_store_memory,\n        )\n        # Register custom serializer for method objects to avoid warning message.\n        # We serialize `MethodType` objects when we use AxisPartition operations.\n        ray.register_custom_serializer(types.MethodType, use_pickle=True)\n\n\nif execution_engine == \"Ray\":\n    initialize_ray()\n    num_cpus = ray.global_state.cluster_resources()[\"CPU\"]\nelif execution_engine == \"Dask\":\n    from distributed.client import _get_global_client\n\n    if threading.current_thread().name == \"MainThread\":\n        # initialize the dask client\n        client = _get_global_client()\n        if client is None:\n            from distributed import Client\n\n            client = Client()\n        num_cpus = sum(client.ncores().values())\nelif execution_engine != \"Python\":\n    raise ImportError(\"Unrecognized execution engine: {}.\".format(execution_engine))\n\nDEFAULT_NPARTITIONS = max(4, int(num_cpus))\n\n__all__ = [\n    \"DataFrame\",\n    \"Series\",\n    \"read_csv\",\n    \"read_parquet\",\n    \"read_json\",\n    \"read_html\",\n    \"read_clipboard\",\n    \"read_excel\",\n    \"read_hdf\",\n    \"read_feather\",\n    \"read_msgpack\",\n    \"read_stata\",\n    \"read_sas\",\n    \"read_pickle\",\n    \"read_sql\",\n    \"read_gbq\",\n    \"read_table\",\n    \"concat\",\n    \"eval\",\n    \"unique\",\n    \"value_counts\",\n    \"cut\",\n    \"to_numeric\",\n    \"factorize\",\n    \"test\",\n    \"qcut\",\n    \"match\",\n    \"to_datetime\",\n    \"get_dummies\",\n    \"isna\",\n    \"isnull\",\n    \"merge\",\n    \"pivot_table\",\n    \"Panel\",\n    \"date_range\",\n    \"Index\",\n    \"MultiIndex\",\n    \"Series\",\n    \"bdate_range\",\n    \"period_range\",\n    \"DatetimeIndex\",\n    \"to_timedelta\",\n    \"set_eng_float_format\",\n    \"set_option\",\n    \"CategoricalIndex\",\n    \"Timedelta\",\n    \"Timestamp\",\n    \"NaT\",\n    \"PeriodIndex\",\n    \"Categorical\",\n    \"__version__\",\n    \"melt\",\n    \"crosstab\",\n    \"plotting\",\n]\n\ndel pandas\n", "path": "modin/pandas/__init__.py"}]}