We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG] read_ray_dataset error when block is pandas DataFrame
<!--
Thank you for your contribution!

Please review https://github.com/mars-project/mars/blob/master/CONTRIBUTING.rst before opening an issue.
-->

**Describe the bug**
 read_ray_dataset error when block is pandas DataFrame.

**To Reproduce**
To help us reproducing this bug, please provide information below:
1. Your Python version: 3.8
2. The version of Mars you use: master
3. Versions of crucial packages, such as numpy, scipy and pandas
4. Full stack of the error.
```
ds = Dataset(num_blocks=1, num_rows=5000, schema={a: int64, b: int64}), columns = None
incremental_index = False, kwargs = {}
refs = [ObjectRef(c2668a65bda616c1ffffffffffffffffffffffff0100000001000000)]

    def read_ray_dataset(ds, columns=None, incremental_index=False, **kwargs):
        assert isinstance(ds, real_ray_dataset.Dataset)
        refs = ds.to_pandas_refs()
>       dtypes = ds.schema().empty_table().to_pandas().dtypes
E       AttributeError: 'PandasBlockSchema' object has no attribute 'empty_table'

../../../../../../opt/anaconda3/lib/python3.8/site-packages/mars/dataframe/datasource/read_raydataset.py:138: AttributeError
```

5. Minimized code to reproduce the error.
```
    @pytest.mark.parametrize("use_mars_api", [True])
    def test_mars(ray_start_regular, use_mars_api):
        import pandas as pd
    
        cluster = mars.new_cluster_in_ray(worker_num=2, worker_cpu=1)
        n = 10000
        pdf = pd.DataFrame({"a": list(range(n)), "b": list(range(n, 2 * n))})
        df = md.DataFrame(pdf)
    
        # Convert mars dataframe to ray dataset
        if use_mars_api:
            ds = md.to_ray_dataset(df)
        else:
            ds = ray.data.from_mars(df)
        pd.testing.assert_frame_equal(ds.to_pandas(), df.to_pandas())
        ds2 = ds.filter(lambda row: row["a"] % 2 == 0)
        assert ds2.take(5) == [{"a": 2 * i, "b": n + 2 * i} for i in range(5)]
    
        # Convert ray dataset to mars dataframe
        if use_mars_api:
>           df2 = md.read_ray_dataset(ds2)

ray/data/tests/test_mars.py:36: 
```

**Expected behavior**
A clear and concise description of what you expected to happen.

**Additional context**
Add any other context about the problem here.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mars/dataframe/datasource/read_raydataset.py`
Content:
```
1 # Copyright 1999-2021 Alibaba Group Holding Ltd.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 import functools
15 import warnings
16 
17 import numpy as np
18 import pandas as pd
19 
20 from ... import opcodes as OperandDef
21 from ...core import OutputType
22 from ...serialization.serializables import (
23     AnyField,
24     BoolField,
25     ListField,
26     Int64Field,
27     ReferenceField,
28 )
29 from ..utils import parse_index, lazy_import, tokenize
30 from .core import (
31     IncrementalIndexDatasource,
32     IncrementalIndexDataSourceMixin,
33     HeadOptimizedDataSource,
34 )
35 
36 ray = lazy_import("ray")
37 # Ray Datasets is available in early preview at ray.data with Ray 1.6+
38 # (and ray.experimental.data in Ray 1.5)
39 ray_dataset = lazy_import("ray.data", rename="ray_dataset")
40 ray_exp_dataset = lazy_import("ray.experimental.data", rename="ray_exp_dataset")
41 real_ray_dataset = ray_dataset or ray_exp_dataset
42 
43 
44 class DataFrameReadRayDataset(
45     IncrementalIndexDatasource, IncrementalIndexDataSourceMixin
46 ):
47     _op_type_ = OperandDef.READ_RAYDATASET
48 
49     refs = AnyField("refs", default=None)
50     columns = ListField("columns", default=None)
51     incremental_index = BoolField("incremental_index", default=None)
52     nrows = Int64Field("nrows", default=None)
53 
54     @classmethod
55     def _tile_partitioned(cls, op: "DataFrameReadRayDataset"):
56         out_df = op.outputs[0]
57         shape = (np.nan, out_df.shape[1])
58         dtypes = out_df.dtypes
59         dataset = op.refs
60 
61         chunk_index = 0
62         out_chunks = []
63         for object_ref in dataset:
64             chunk_op = op.copy().reset_key()
65             chunk_op._refs = [object_ref]
66             new_chunk = chunk_op.new_chunk(
67                 None,
68                 shape=shape,
69                 index=(chunk_index, 0),
70                 index_value=out_df.index_value,
71                 columns_value=out_df.columns_value,
72                 dtypes=dtypes,
73             )
74             out_chunks.append(new_chunk)
75             chunk_index += 1
76 
77         new_op = op.copy()
78         nsplits = ((np.nan,) * len(out_chunks), (out_df.shape[1],))
79         return new_op.new_dataframes(
80             None,
81             out_df.shape,
82             dtypes=dtypes,
83             index_value=out_df.index_value,
84             columns_value=out_df.columns_value,
85             chunks=out_chunks,
86             nsplits=nsplits,
87         )
88 
89     @classmethod
90     def _tile(cls, op):
91         return cls._tile_partitioned(op)
92 
93     @classmethod
94     def execute(cls, ctx, op: "DataFrameReadRayDataset"):
95         out = op.outputs[0]
96         ref = op.refs[0]
97 
98         df = ray.get(ref)
99         ctx[out.key] = df
100 
101     def __call__(self, index_value=None, columns_value=None, dtypes=None):
102         shape = (np.nan, len(dtypes))
103         return self.new_dataframe(
104             None,
105             shape,
106             dtypes=dtypes,
107             index_value=index_value,
108             columns_value=columns_value,
109         )
110 
111 
112 def read_ray_dataset(ds, columns=None, incremental_index=False, **kwargs):
113     assert isinstance(ds, real_ray_dataset.Dataset)
114     refs = ds.to_pandas_refs()
115     dtypes = ds.schema().empty_table().to_pandas().dtypes
116     index_value = parse_index(pd.RangeIndex(-1))
117     columns_value = parse_index(dtypes.index, store_data=True)
118 
119     op = DataFrameReadRayDataset(
120         refs=refs, columns=columns, incremental_index=incremental_index
121     )
122     return op(index_value=index_value, columns_value=columns_value, dtypes=dtypes)
123 
124 
125 # keep it for back compatibility
126 @functools.wraps(read_ray_dataset)
127 def read_raydataset(*args, **kwargs):
128     warnings.warn(
129         "read_raydataset has been renamed to read_ray_dataset",
130         DeprecationWarning,
131     )
132     return read_ray_dataset(*args, **kwargs)
133 
134 
135 class DataFrameReadMLDataset(HeadOptimizedDataSource):
136     _op_type_ = OperandDef.READ_MLDATASET
137 
138     mldataset = ReferenceField("mldataset", "ray.util.data.MLDataset", default=None)
139     columns = ListField("columns", default=None)
140 
141     def __init__(self, **kw):
142         super().__init__(_output_types=[OutputType.dataframe], **kw)
143 
144     def _update_key(self):
145         """We can't direct generate token for mldataset when we use
146         ray client, so we use all mldataset's actor_id to generate
147         token.
148         """
149         datas = []
150         for value in self._values_:
151             if isinstance(value, ray.util.data.MLDataset):
152                 actor_sets = [
153                     ([str(actor) for actor in actor_set.actors], actor_set.transforms)
154                     for actor_set in value.actor_sets
155                 ]
156                 datas.append(actor_sets)
157                 continue
158             datas.append(value)
159         self._obj_set("_key", tokenize(type(self).__name__, *datas))
160         return self
161 
162     def __call__(self, dtypes, nrows: int):
163         columns_value = parse_index(dtypes.index, store_data=True)
164         index_value = parse_index(pd.RangeIndex(nrows))
165         return self.new_dataframe(
166             None,
167             (nrows, len(dtypes)),
168             dtypes=dtypes,
169             index_value=index_value,
170             columns_value=columns_value,
171         )
172 
173     @classmethod
174     def tile(cls, op: "DataFrameReadMLDataset"):
175         count_iter = op.mldataset.for_each(lambda df: len(df))
176         nsplits = [sum(shard) for shard in count_iter.shards()]
177         nsplits_acc = np.cumsum(nsplits)
178         out_df = op.outputs[0]
179         out_chunks = []
180         for shard_index in range(op.mldataset.num_shards()):
181             chunk_op = op.copy().reset_key()
182             # Make chunk key unique, otherwise all chunk will have same key.
183             # See `DataFrameFromRecords#tile`
184             chunk_op.extra_params["shard_index"] = shard_index
185             shape = (nsplits[shard_index], out_df.shape[1])
186             begin_index = nsplits_acc[shard_index] - nsplits[shard_index]
187             end_index = nsplits_acc[shard_index]
188             index = parse_index(pd.RangeIndex(start=begin_index, stop=end_index))
189             new_chunk = chunk_op.new_chunk(
190                 None,
191                 shape=shape,
192                 index=(shard_index, 0),
193                 index_value=index,
194                 columns_value=out_df.columns_value,
195                 dtypes=out_df.dtypes,
196             )
197             out_chunks.append(new_chunk)
198         new_op = op.copy()
199         nsplits = ((np.nan,) * len(out_chunks), (out_df.shape[1],))
200         return new_op.new_dataframes(
201             None,
202             out_df.shape,
203             dtypes=out_df.dtypes,
204             index_value=out_df.index_value,
205             columns_value=out_df.columns_value,
206             chunks=out_chunks,
207             nsplits=nsplits,
208         )
209 
210     @classmethod
211     def execute(cls, ctx, op):
212         chunk = op.outputs[0]
213         shard = op.mldataset.get_shard(chunk.index[0])
214         pd_dfs = list(shard)
215         pd_df = pd.concat(pd_dfs).set_index(chunk.index_value.to_pandas())
216         ctx[chunk.key] = pd_df
217 
218 
219 def read_ray_mldataset(mldataset, **kwargs):
220     import ray.util.data
221 
222     assert isinstance(mldataset, ray.util.data.MLDataset)
223     not_empty_dfs = mldataset.filter(lambda df: len(df) > 0).take(1)
224     if not not_empty_dfs:
225         raise ValueError(
226             f"MLDataset {mldataset} is empty, please provide an non-empty dataset."
227         )
228     df_record: pd.DataFrame = not_empty_dfs[0]
229     columns = df_record.columns.names
230     nrows = sum(mldataset.for_each(lambda df: len(df)).gather_async())
231     op = DataFrameReadMLDataset(mldataset=mldataset, columns=columns, nrows=nrows)
232     return op(df_record.dtypes, nrows)
233
```

Path: `mars/services/task/execution/ray/config.py`
Content:
```
1 # Copyright 1999-2021 Alibaba Group Holding Ltd.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 from typing import Dict, List
16 from .....resource import Resource
17 from ..api import ExecutionConfig, register_config_cls
18 from ..utils import get_band_resources_from_config
19 
20 # The default interval seconds to update progress and collect garbage.
21 DEFAULT_SUBTASK_MONITOR_INTERVAL = 1
22 
23 
24 @register_config_cls
25 class RayExecutionConfig(ExecutionConfig):
26     name = "ray"
27 
28     def __init__(self, execution_config: Dict):
29         super().__init__(execution_config)
30         self._ray_execution_config = execution_config[self.backend]
31 
32     def get_band_resources(self):
33         """
34         Get the band resources from config for generating ray virtual
35         resources.
36         """
37         return get_band_resources_from_config(self._ray_execution_config)
38 
39     def get_deploy_band_resources(self) -> List[Dict[str, Resource]]:
40         return []
41 
42     def get_subtask_max_retries(self):
43         return self._ray_execution_config.get("subtask_max_retries")
44 
45     def get_n_cpu(self):
46         return self._ray_execution_config["n_cpu"]
47 
48     def get_n_worker(self):
49         return self._ray_execution_config["n_worker"]
50 
51     def get_subtask_cancel_timeout(self):
52         return self._ray_execution_config.get("subtask_cancel_timeout")
53 
54     def create_task_state_actor_as_needed(self):
55         # Whether create RayTaskState actor as needed.
56         #   - True (default):
57         #     Create RayTaskState actor only when create_remote_object is called.
58         #   - False:
59         #     Create RayTaskState actor in advance when the RayTaskExecutor is created.
60         return self._ray_execution_config.get("create_task_state_actor_as_needed", True)
61 
62     def get_subtask_monitor_interval(self):
63         """
64         The interval seconds for the monitor task to update progress and
65         collect garbage.
66         """
67         return self._ray_execution_config.get(
68             "subtask_monitor_interval", DEFAULT_SUBTASK_MONITOR_INTERVAL
69         )
70
```
--- 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/mars/dataframe/datasource/read_raydataset.py b/mars/dataframe/datasource/read_raydataset.py
--- a/mars/dataframe/datasource/read_raydataset.py
+++ b/mars/dataframe/datasource/read_raydataset.py
@@ -112,10 +112,23 @@
 def read_ray_dataset(ds, columns=None, incremental_index=False, **kwargs):
     assert isinstance(ds, real_ray_dataset.Dataset)
     refs = ds.to_pandas_refs()
-    dtypes = ds.schema().empty_table().to_pandas().dtypes
+    schema = ds.schema()
+
+    import pyarrow as pa
+
+    try:
+        from ray.data.impl.pandas_block import PandasBlockSchema
+
+        if isinstance(schema, PandasBlockSchema):
+            dtypes = pd.Series(schema.types, index=schema.names)
+        elif isinstance(schema, pa.Schema):
+            dtypes = schema.empty_table().to_pandas().dtypes
+        else:
+            raise NotImplementedError(f"Unsupported format of schema {schema}")
+    except ImportError:  # pragma: no cover
+        dtypes = schema.empty_table().to_pandas().dtypes
     index_value = parse_index(pd.RangeIndex(-1))
     columns_value = parse_index(dtypes.index, store_data=True)
-
     op = DataFrameReadRayDataset(
         refs=refs, columns=columns, incremental_index=incremental_index
     )
diff --git a/mars/services/task/execution/ray/config.py b/mars/services/task/execution/ray/config.py
--- a/mars/services/task/execution/ray/config.py
+++ b/mars/services/task/execution/ray/config.py
@@ -40,7 +40,7 @@
         return []
 
     def get_subtask_max_retries(self):
-        return self._ray_execution_config.get("subtask_max_retries")
+        return self._ray_execution_config["subtask_max_retries"]
 
     def get_n_cpu(self):
         return self._ray_execution_config["n_cpu"]
@@ -49,7 +49,7 @@
         return self._ray_execution_config["n_worker"]
 
     def get_subtask_cancel_timeout(self):
-        return self._ray_execution_config.get("subtask_cancel_timeout")
+        return self._ray_execution_config["subtask_cancel_timeout"]
 
     def create_task_state_actor_as_needed(self):
         # Whether create RayTaskState actor as needed.

{"golden_diff": "diff --git a/mars/dataframe/datasource/read_raydataset.py b/mars/dataframe/datasource/read_raydataset.py\n--- a/mars/dataframe/datasource/read_raydataset.py\n+++ b/mars/dataframe/datasource/read_raydataset.py\n@@ -112,10 +112,23 @@\n def read_ray_dataset(ds, columns=None, incremental_index=False, **kwargs):\n     assert isinstance(ds, real_ray_dataset.Dataset)\n     refs = ds.to_pandas_refs()\n-    dtypes = ds.schema().empty_table().to_pandas().dtypes\n+    schema = ds.schema()\n+\n+    import pyarrow as pa\n+\n+    try:\n+        from ray.data.impl.pandas_block import PandasBlockSchema\n+\n+        if isinstance(schema, PandasBlockSchema):\n+            dtypes = pd.Series(schema.types, index=schema.names)\n+        elif isinstance(schema, pa.Schema):\n+            dtypes = schema.empty_table().to_pandas().dtypes\n+        else:\n+            raise NotImplementedError(f\"Unsupported format of schema {schema}\")\n+    except ImportError:  # pragma: no cover\n+        dtypes = schema.empty_table().to_pandas().dtypes\n     index_value = parse_index(pd.RangeIndex(-1))\n     columns_value = parse_index(dtypes.index, store_data=True)\n-\n     op = DataFrameReadRayDataset(\n         refs=refs, columns=columns, incremental_index=incremental_index\n     )\ndiff --git a/mars/services/task/execution/ray/config.py b/mars/services/task/execution/ray/config.py\n--- a/mars/services/task/execution/ray/config.py\n+++ b/mars/services/task/execution/ray/config.py\n@@ -40,7 +40,7 @@\n         return []\n \n     def get_subtask_max_retries(self):\n-        return self._ray_execution_config.get(\"subtask_max_retries\")\n+        return self._ray_execution_config[\"subtask_max_retries\"]\n \n     def get_n_cpu(self):\n         return self._ray_execution_config[\"n_cpu\"]\n@@ -49,7 +49,7 @@\n         return self._ray_execution_config[\"n_worker\"]\n \n     def get_subtask_cancel_timeout(self):\n-        return self._ray_execution_config.get(\"subtask_cancel_timeout\")\n+        return self._ray_execution_config[\"subtask_cancel_timeout\"]\n \n     def create_task_state_actor_as_needed(self):\n         # Whether create RayTaskState actor as needed.\n", "issue": "[BUG] read_ray_dataset error when block is pandas DataFrame\n<!--\r\nThank you for your contribution!\r\n\r\nPlease review https://github.com/mars-project/mars/blob/master/CONTRIBUTING.rst before opening an issue.\r\n-->\r\n\r\n**Describe the bug**\r\n read_ray_dataset error when block is pandas DataFrame.\r\n\r\n**To Reproduce**\r\nTo help us reproducing this bug, please provide information below:\r\n1. Your Python version: 3.8\r\n2. The version of Mars you use: master\r\n3. Versions of crucial packages, such as numpy, scipy and pandas\r\n4. Full stack of the error.\r\n```\r\nds = Dataset(num_blocks=1, num_rows=5000, schema={a: int64, b: int64}), columns = None\r\nincremental_index = False, kwargs = {}\r\nrefs = [ObjectRef(c2668a65bda616c1ffffffffffffffffffffffff0100000001000000)]\r\n\r\n    def read_ray_dataset(ds, columns=None, incremental_index=False, **kwargs):\r\n        assert isinstance(ds, real_ray_dataset.Dataset)\r\n        refs = ds.to_pandas_refs()\r\n>       dtypes = ds.schema().empty_table().to_pandas().dtypes\r\nE       AttributeError: 'PandasBlockSchema' object has no attribute 'empty_table'\r\n\r\n../../../../../../opt/anaconda3/lib/python3.8/site-packages/mars/dataframe/datasource/read_raydataset.py:138: AttributeError\r\n```\r\n\r\n5. Minimized code to reproduce the error.\r\n```\r\n    @pytest.mark.parametrize(\"use_mars_api\", [True])\r\n    def test_mars(ray_start_regular, use_mars_api):\r\n        import pandas as pd\r\n    \r\n        cluster = mars.new_cluster_in_ray(worker_num=2, worker_cpu=1)\r\n        n = 10000\r\n        pdf = pd.DataFrame({\"a\": list(range(n)), \"b\": list(range(n, 2 * n))})\r\n        df = md.DataFrame(pdf)\r\n    \r\n        # Convert mars dataframe to ray dataset\r\n        if use_mars_api:\r\n            ds = md.to_ray_dataset(df)\r\n        else:\r\n            ds = ray.data.from_mars(df)\r\n        pd.testing.assert_frame_equal(ds.to_pandas(), df.to_pandas())\r\n        ds2 = ds.filter(lambda row: row[\"a\"] % 2 == 0)\r\n        assert ds2.take(5) == [{\"a\": 2 * i, \"b\": n + 2 * i} for i in range(5)]\r\n    \r\n        # Convert ray dataset to mars dataframe\r\n        if use_mars_api:\r\n>           df2 = md.read_ray_dataset(ds2)\r\n\r\nray/data/tests/test_mars.py:36: \r\n```\r\n\r\n**Expected behavior**\r\nA clear and concise description of what you expected to happen.\r\n\r\n**Additional context**\r\nAdd any other context about the problem here.\r\n\n", "before_files": [{"content": "# Copyright 1999-2021 Alibaba Group Holding Ltd.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nimport functools\nimport warnings\n\nimport numpy as np\nimport pandas as pd\n\nfrom ... import opcodes as OperandDef\nfrom ...core import OutputType\nfrom ...serialization.serializables import (\n    AnyField,\n    BoolField,\n    ListField,\n    Int64Field,\n    ReferenceField,\n)\nfrom ..utils import parse_index, lazy_import, tokenize\nfrom .core import (\n    IncrementalIndexDatasource,\n    IncrementalIndexDataSourceMixin,\n    HeadOptimizedDataSource,\n)\n\nray = lazy_import(\"ray\")\n# Ray Datasets is available in early preview at ray.data with Ray 1.6+\n# (and ray.experimental.data in Ray 1.5)\nray_dataset = lazy_import(\"ray.data\", rename=\"ray_dataset\")\nray_exp_dataset = lazy_import(\"ray.experimental.data\", rename=\"ray_exp_dataset\")\nreal_ray_dataset = ray_dataset or ray_exp_dataset\n\n\nclass DataFrameReadRayDataset(\n    IncrementalIndexDatasource, IncrementalIndexDataSourceMixin\n):\n    _op_type_ = OperandDef.READ_RAYDATASET\n\n    refs = AnyField(\"refs\", default=None)\n    columns = ListField(\"columns\", default=None)\n    incremental_index = BoolField(\"incremental_index\", default=None)\n    nrows = Int64Field(\"nrows\", default=None)\n\n    @classmethod\n    def _tile_partitioned(cls, op: \"DataFrameReadRayDataset\"):\n        out_df = op.outputs[0]\n        shape = (np.nan, out_df.shape[1])\n        dtypes = out_df.dtypes\n        dataset = op.refs\n\n        chunk_index = 0\n        out_chunks = []\n        for object_ref in dataset:\n            chunk_op = op.copy().reset_key()\n            chunk_op._refs = [object_ref]\n            new_chunk = chunk_op.new_chunk(\n                None,\n                shape=shape,\n                index=(chunk_index, 0),\n                index_value=out_df.index_value,\n                columns_value=out_df.columns_value,\n                dtypes=dtypes,\n            )\n            out_chunks.append(new_chunk)\n            chunk_index += 1\n\n        new_op = op.copy()\n        nsplits = ((np.nan,) * len(out_chunks), (out_df.shape[1],))\n        return new_op.new_dataframes(\n            None,\n            out_df.shape,\n            dtypes=dtypes,\n            index_value=out_df.index_value,\n            columns_value=out_df.columns_value,\n            chunks=out_chunks,\n            nsplits=nsplits,\n        )\n\n    @classmethod\n    def _tile(cls, op):\n        return cls._tile_partitioned(op)\n\n    @classmethod\n    def execute(cls, ctx, op: \"DataFrameReadRayDataset\"):\n        out = op.outputs[0]\n        ref = op.refs[0]\n\n        df = ray.get(ref)\n        ctx[out.key] = df\n\n    def __call__(self, index_value=None, columns_value=None, dtypes=None):\n        shape = (np.nan, len(dtypes))\n        return self.new_dataframe(\n            None,\n            shape,\n            dtypes=dtypes,\n            index_value=index_value,\n            columns_value=columns_value,\n        )\n\n\ndef read_ray_dataset(ds, columns=None, incremental_index=False, **kwargs):\n    assert isinstance(ds, real_ray_dataset.Dataset)\n    refs = ds.to_pandas_refs()\n    dtypes = ds.schema().empty_table().to_pandas().dtypes\n    index_value = parse_index(pd.RangeIndex(-1))\n    columns_value = parse_index(dtypes.index, store_data=True)\n\n    op = DataFrameReadRayDataset(\n        refs=refs, columns=columns, incremental_index=incremental_index\n    )\n    return op(index_value=index_value, columns_value=columns_value, dtypes=dtypes)\n\n\n# keep it for back compatibility\n@functools.wraps(read_ray_dataset)\ndef read_raydataset(*args, **kwargs):\n    warnings.warn(\n        \"read_raydataset has been renamed to read_ray_dataset\",\n        DeprecationWarning,\n    )\n    return read_ray_dataset(*args, **kwargs)\n\n\nclass DataFrameReadMLDataset(HeadOptimizedDataSource):\n    _op_type_ = OperandDef.READ_MLDATASET\n\n    mldataset = ReferenceField(\"mldataset\", \"ray.util.data.MLDataset\", default=None)\n    columns = ListField(\"columns\", default=None)\n\n    def __init__(self, **kw):\n        super().__init__(_output_types=[OutputType.dataframe], **kw)\n\n    def _update_key(self):\n        \"\"\"We can't direct generate token for mldataset when we use\n        ray client, so we use all mldataset's actor_id to generate\n        token.\n        \"\"\"\n        datas = []\n        for value in self._values_:\n            if isinstance(value, ray.util.data.MLDataset):\n                actor_sets = [\n                    ([str(actor) for actor in actor_set.actors], actor_set.transforms)\n                    for actor_set in value.actor_sets\n                ]\n                datas.append(actor_sets)\n                continue\n            datas.append(value)\n        self._obj_set(\"_key\", tokenize(type(self).__name__, *datas))\n        return self\n\n    def __call__(self, dtypes, nrows: int):\n        columns_value = parse_index(dtypes.index, store_data=True)\n        index_value = parse_index(pd.RangeIndex(nrows))\n        return self.new_dataframe(\n            None,\n            (nrows, len(dtypes)),\n            dtypes=dtypes,\n            index_value=index_value,\n            columns_value=columns_value,\n        )\n\n    @classmethod\n    def tile(cls, op: \"DataFrameReadMLDataset\"):\n        count_iter = op.mldataset.for_each(lambda df: len(df))\n        nsplits = [sum(shard) for shard in count_iter.shards()]\n        nsplits_acc = np.cumsum(nsplits)\n        out_df = op.outputs[0]\n        out_chunks = []\n        for shard_index in range(op.mldataset.num_shards()):\n            chunk_op = op.copy().reset_key()\n            # Make chunk key unique, otherwise all chunk will have same key.\n            # See `DataFrameFromRecords#tile`\n            chunk_op.extra_params[\"shard_index\"] = shard_index\n            shape = (nsplits[shard_index], out_df.shape[1])\n            begin_index = nsplits_acc[shard_index] - nsplits[shard_index]\n            end_index = nsplits_acc[shard_index]\n            index = parse_index(pd.RangeIndex(start=begin_index, stop=end_index))\n            new_chunk = chunk_op.new_chunk(\n                None,\n                shape=shape,\n                index=(shard_index, 0),\n                index_value=index,\n                columns_value=out_df.columns_value,\n                dtypes=out_df.dtypes,\n            )\n            out_chunks.append(new_chunk)\n        new_op = op.copy()\n        nsplits = ((np.nan,) * len(out_chunks), (out_df.shape[1],))\n        return new_op.new_dataframes(\n            None,\n            out_df.shape,\n            dtypes=out_df.dtypes,\n            index_value=out_df.index_value,\n            columns_value=out_df.columns_value,\n            chunks=out_chunks,\n            nsplits=nsplits,\n        )\n\n    @classmethod\n    def execute(cls, ctx, op):\n        chunk = op.outputs[0]\n        shard = op.mldataset.get_shard(chunk.index[0])\n        pd_dfs = list(shard)\n        pd_df = pd.concat(pd_dfs).set_index(chunk.index_value.to_pandas())\n        ctx[chunk.key] = pd_df\n\n\ndef read_ray_mldataset(mldataset, **kwargs):\n    import ray.util.data\n\n    assert isinstance(mldataset, ray.util.data.MLDataset)\n    not_empty_dfs = mldataset.filter(lambda df: len(df) > 0).take(1)\n    if not not_empty_dfs:\n        raise ValueError(\n            f\"MLDataset {mldataset} is empty, please provide an non-empty dataset.\"\n        )\n    df_record: pd.DataFrame = not_empty_dfs[0]\n    columns = df_record.columns.names\n    nrows = sum(mldataset.for_each(lambda df: len(df)).gather_async())\n    op = DataFrameReadMLDataset(mldataset=mldataset, columns=columns, nrows=nrows)\n    return op(df_record.dtypes, nrows)\n", "path": "mars/dataframe/datasource/read_raydataset.py"}, {"content": "# Copyright 1999-2021 Alibaba Group Holding Ltd.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom typing import Dict, List\nfrom .....resource import Resource\nfrom ..api import ExecutionConfig, register_config_cls\nfrom ..utils import get_band_resources_from_config\n\n# The default interval seconds to update progress and collect garbage.\nDEFAULT_SUBTASK_MONITOR_INTERVAL = 1\n\n\n@register_config_cls\nclass RayExecutionConfig(ExecutionConfig):\n    name = \"ray\"\n\n    def __init__(self, execution_config: Dict):\n        super().__init__(execution_config)\n        self._ray_execution_config = execution_config[self.backend]\n\n    def get_band_resources(self):\n        \"\"\"\n        Get the band resources from config for generating ray virtual\n        resources.\n        \"\"\"\n        return get_band_resources_from_config(self._ray_execution_config)\n\n    def get_deploy_band_resources(self) -> List[Dict[str, Resource]]:\n        return []\n\n    def get_subtask_max_retries(self):\n        return self._ray_execution_config.get(\"subtask_max_retries\")\n\n    def get_n_cpu(self):\n        return self._ray_execution_config[\"n_cpu\"]\n\n    def get_n_worker(self):\n        return self._ray_execution_config[\"n_worker\"]\n\n    def get_subtask_cancel_timeout(self):\n        return self._ray_execution_config.get(\"subtask_cancel_timeout\")\n\n    def create_task_state_actor_as_needed(self):\n        # Whether create RayTaskState actor as needed.\n        #   - True (default):\n        #     Create RayTaskState actor only when create_remote_object is called.\n        #   - False:\n        #     Create RayTaskState actor in advance when the RayTaskExecutor is created.\n        return self._ray_execution_config.get(\"create_task_state_actor_as_needed\", True)\n\n    def get_subtask_monitor_interval(self):\n        \"\"\"\n        The interval seconds for the monitor task to update progress and\n        collect garbage.\n        \"\"\"\n        return self._ray_execution_config.get(\n            \"subtask_monitor_interval\", DEFAULT_SUBTASK_MONITOR_INTERVAL\n        )\n", "path": "mars/services/task/execution/ray/config.py"}], "after_files": [{"content": "# Copyright 1999-2021 Alibaba Group Holding Ltd.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nimport functools\nimport warnings\n\nimport numpy as np\nimport pandas as pd\n\nfrom ... import opcodes as OperandDef\nfrom ...core import OutputType\nfrom ...serialization.serializables import (\n    AnyField,\n    BoolField,\n    ListField,\n    Int64Field,\n    ReferenceField,\n)\nfrom ..utils import parse_index, lazy_import, tokenize\nfrom .core import (\n    IncrementalIndexDatasource,\n    IncrementalIndexDataSourceMixin,\n    HeadOptimizedDataSource,\n)\n\nray = lazy_import(\"ray\")\n# Ray Datasets is available in early preview at ray.data with Ray 1.6+\n# (and ray.experimental.data in Ray 1.5)\nray_dataset = lazy_import(\"ray.data\", rename=\"ray_dataset\")\nray_exp_dataset = lazy_import(\"ray.experimental.data\", rename=\"ray_exp_dataset\")\nreal_ray_dataset = ray_dataset or ray_exp_dataset\n\n\nclass DataFrameReadRayDataset(\n    IncrementalIndexDatasource, IncrementalIndexDataSourceMixin\n):\n    _op_type_ = OperandDef.READ_RAYDATASET\n\n    refs = AnyField(\"refs\", default=None)\n    columns = ListField(\"columns\", default=None)\n    incremental_index = BoolField(\"incremental_index\", default=None)\n    nrows = Int64Field(\"nrows\", default=None)\n\n    @classmethod\n    def _tile_partitioned(cls, op: \"DataFrameReadRayDataset\"):\n        out_df = op.outputs[0]\n        shape = (np.nan, out_df.shape[1])\n        dtypes = out_df.dtypes\n        dataset = op.refs\n\n        chunk_index = 0\n        out_chunks = []\n        for object_ref in dataset:\n            chunk_op = op.copy().reset_key()\n            chunk_op._refs = [object_ref]\n            new_chunk = chunk_op.new_chunk(\n                None,\n                shape=shape,\n                index=(chunk_index, 0),\n                index_value=out_df.index_value,\n                columns_value=out_df.columns_value,\n                dtypes=dtypes,\n            )\n            out_chunks.append(new_chunk)\n            chunk_index += 1\n\n        new_op = op.copy()\n        nsplits = ((np.nan,) * len(out_chunks), (out_df.shape[1],))\n        return new_op.new_dataframes(\n            None,\n            out_df.shape,\n            dtypes=dtypes,\n            index_value=out_df.index_value,\n            columns_value=out_df.columns_value,\n            chunks=out_chunks,\n            nsplits=nsplits,\n        )\n\n    @classmethod\n    def _tile(cls, op):\n        return cls._tile_partitioned(op)\n\n    @classmethod\n    def execute(cls, ctx, op: \"DataFrameReadRayDataset\"):\n        out = op.outputs[0]\n        ref = op.refs[0]\n\n        df = ray.get(ref)\n        ctx[out.key] = df\n\n    def __call__(self, index_value=None, columns_value=None, dtypes=None):\n        shape = (np.nan, len(dtypes))\n        return self.new_dataframe(\n            None,\n            shape,\n            dtypes=dtypes,\n            index_value=index_value,\n            columns_value=columns_value,\n        )\n\n\ndef read_ray_dataset(ds, columns=None, incremental_index=False, **kwargs):\n    assert isinstance(ds, real_ray_dataset.Dataset)\n    refs = ds.to_pandas_refs()\n    schema = ds.schema()\n\n    import pyarrow as pa\n\n    try:\n        from ray.data.impl.pandas_block import PandasBlockSchema\n\n        if isinstance(schema, PandasBlockSchema):\n            dtypes = pd.Series(schema.types, index=schema.names)\n        elif isinstance(schema, pa.Schema):\n            dtypes = schema.empty_table().to_pandas().dtypes\n        else:\n            raise NotImplementedError(f\"Unsupported format of schema {schema}\")\n    except ImportError:  # pragma: no cover\n        dtypes = schema.empty_table().to_pandas().dtypes\n    index_value = parse_index(pd.RangeIndex(-1))\n    columns_value = parse_index(dtypes.index, store_data=True)\n    op = DataFrameReadRayDataset(\n        refs=refs, columns=columns, incremental_index=incremental_index\n    )\n    return op(index_value=index_value, columns_value=columns_value, dtypes=dtypes)\n\n\n# keep it for back compatibility\n@functools.wraps(read_ray_dataset)\ndef read_raydataset(*args, **kwargs):\n    warnings.warn(\n        \"read_raydataset has been renamed to read_ray_dataset\",\n        DeprecationWarning,\n    )\n    return read_ray_dataset(*args, **kwargs)\n\n\nclass DataFrameReadMLDataset(HeadOptimizedDataSource):\n    _op_type_ = OperandDef.READ_MLDATASET\n\n    mldataset = ReferenceField(\"mldataset\", \"ray.util.data.MLDataset\", default=None)\n    columns = ListField(\"columns\", default=None)\n\n    def __init__(self, **kw):\n        super().__init__(_output_types=[OutputType.dataframe], **kw)\n\n    def _update_key(self):\n        \"\"\"We can't direct generate token for mldataset when we use\n        ray client, so we use all mldataset's actor_id to generate\n        token.\n        \"\"\"\n        datas = []\n        for value in self._values_:\n            if isinstance(value, ray.util.data.MLDataset):\n                actor_sets = [\n                    ([str(actor) for actor in actor_set.actors], actor_set.transforms)\n                    for actor_set in value.actor_sets\n                ]\n                datas.append(actor_sets)\n                continue\n            datas.append(value)\n        self._obj_set(\"_key\", tokenize(type(self).__name__, *datas))\n        return self\n\n    def __call__(self, dtypes, nrows: int):\n        columns_value = parse_index(dtypes.index, store_data=True)\n        index_value = parse_index(pd.RangeIndex(nrows))\n        return self.new_dataframe(\n            None,\n            (nrows, len(dtypes)),\n            dtypes=dtypes,\n            index_value=index_value,\n            columns_value=columns_value,\n        )\n\n    @classmethod\n    def tile(cls, op: \"DataFrameReadMLDataset\"):\n        count_iter = op.mldataset.for_each(lambda df: len(df))\n        nsplits = [sum(shard) for shard in count_iter.shards()]\n        nsplits_acc = np.cumsum(nsplits)\n        out_df = op.outputs[0]\n        out_chunks = []\n        for shard_index in range(op.mldataset.num_shards()):\n            chunk_op = op.copy().reset_key()\n            # Make chunk key unique, otherwise all chunk will have same key.\n            # See `DataFrameFromRecords#tile`\n            chunk_op.extra_params[\"shard_index\"] = shard_index\n            shape = (nsplits[shard_index], out_df.shape[1])\n            begin_index = nsplits_acc[shard_index] - nsplits[shard_index]\n            end_index = nsplits_acc[shard_index]\n            index = parse_index(pd.RangeIndex(start=begin_index, stop=end_index))\n            new_chunk = chunk_op.new_chunk(\n                None,\n                shape=shape,\n                index=(shard_index, 0),\n                index_value=index,\n                columns_value=out_df.columns_value,\n                dtypes=out_df.dtypes,\n            )\n            out_chunks.append(new_chunk)\n        new_op = op.copy()\n        nsplits = ((np.nan,) * len(out_chunks), (out_df.shape[1],))\n        return new_op.new_dataframes(\n            None,\n            out_df.shape,\n            dtypes=out_df.dtypes,\n            index_value=out_df.index_value,\n            columns_value=out_df.columns_value,\n            chunks=out_chunks,\n            nsplits=nsplits,\n        )\n\n    @classmethod\n    def execute(cls, ctx, op):\n        chunk = op.outputs[0]\n        shard = op.mldataset.get_shard(chunk.index[0])\n        pd_dfs = list(shard)\n        pd_df = pd.concat(pd_dfs).set_index(chunk.index_value.to_pandas())\n        ctx[chunk.key] = pd_df\n\n\ndef read_ray_mldataset(mldataset, **kwargs):\n    import ray.util.data\n\n    assert isinstance(mldataset, ray.util.data.MLDataset)\n    not_empty_dfs = mldataset.filter(lambda df: len(df) > 0).take(1)\n    if not not_empty_dfs:\n        raise ValueError(\n            f\"MLDataset {mldataset} is empty, please provide an non-empty dataset.\"\n        )\n    df_record: pd.DataFrame = not_empty_dfs[0]\n    columns = df_record.columns.names\n    nrows = sum(mldataset.for_each(lambda df: len(df)).gather_async())\n    op = DataFrameReadMLDataset(mldataset=mldataset, columns=columns, nrows=nrows)\n    return op(df_record.dtypes, nrows)\n", "path": "mars/dataframe/datasource/read_raydataset.py"}, {"content": "# Copyright 1999-2021 Alibaba Group Holding Ltd.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nfrom typing import Dict, List\nfrom .....resource import Resource\nfrom ..api import ExecutionConfig, register_config_cls\nfrom ..utils import get_band_resources_from_config\n\n# The default interval seconds to update progress and collect garbage.\nDEFAULT_SUBTASK_MONITOR_INTERVAL = 1\n\n\n@register_config_cls\nclass RayExecutionConfig(ExecutionConfig):\n    name = \"ray\"\n\n    def __init__(self, execution_config: Dict):\n        super().__init__(execution_config)\n        self._ray_execution_config = execution_config[self.backend]\n\n    def get_band_resources(self):\n        \"\"\"\n        Get the band resources from config for generating ray virtual\n        resources.\n        \"\"\"\n        return get_band_resources_from_config(self._ray_execution_config)\n\n    def get_deploy_band_resources(self) -> List[Dict[str, Resource]]:\n        return []\n\n    def get_subtask_max_retries(self):\n        return self._ray_execution_config[\"subtask_max_retries\"]\n\n    def get_n_cpu(self):\n        return self._ray_execution_config[\"n_cpu\"]\n\n    def get_n_worker(self):\n        return self._ray_execution_config[\"n_worker\"]\n\n    def get_subtask_cancel_timeout(self):\n        return self._ray_execution_config[\"subtask_cancel_timeout\"]\n\n    def create_task_state_actor_as_needed(self):\n        # Whether create RayTaskState actor as needed.\n        #   - True (default):\n        #     Create RayTaskState actor only when create_remote_object is called.\n        #   - False:\n        #     Create RayTaskState actor in advance when the RayTaskExecutor is created.\n        return self._ray_execution_config.get(\"create_task_state_actor_as_needed\", True)\n\n    def get_subtask_monitor_interval(self):\n        \"\"\"\n        The interval seconds for the monitor task to update progress and\n        collect garbage.\n        \"\"\"\n        return self._ray_execution_config.get(\n            \"subtask_monitor_interval\", DEFAULT_SUBTASK_MONITOR_INTERVAL\n        )\n", "path": "mars/services/task/execution/ray/config.py"}]}