Make clearer separation of sessions and allow customized implementations (#3335)

Author: wjsi

benchmarks/README.md

@@ -48,7 +48,7 @@ This script assumes `tpch-dbgen` is in the same directory. If you downloaded it
 
 ### Installation
 
-Follow the intstructions [here](https://docs.pymars.org/en/latest/installation/index.html).
+Follow the instructions [here](https://docs.pymars.org/en/latest/installation/index.html).
 
 ### Running queries
 

mars/contrib/dask/converter.py

@@ -15,9 +15,9 @@
 from dask import is_dask_collection, optimize
 from dask.bag import Bag
 
+from ...remote import spawn
 from .scheduler import mars_dask_get
 from .utils import reduce
-from ...remote import spawn
 
 
 def convert_dask_collection(dc):

mars/contrib/dask/scheduler.py

@@ -12,12 +12,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import List, Tuple, Union
+
 from dask.core import istask, ishashable
 
-from typing import List, Tuple, Union
-from .utils import reduce
 from ...remote import spawn
-from ...deploy.oscar.session import execute
+from ...session import execute
+from .utils import reduce
 
 
 def mars_scheduler(dsk: dict, keys: Union[List[List[str]], List[str]]):

mars/core/entity/executable.py

@@ -38,8 +38,8 @@ def start(self):
         self._decref_thread.start()
 
     def _thread_body(self):
-        from ...deploy.oscar.session import SyncSession
         from ...oscar.errors import ActorNotExist
+        from ...session import SyncSession
 
         while True:
             key, session_ref, fut = self._queue.get()
@@ -130,7 +130,7 @@ def register(self, tileable: TileableType, session: SessionType):
 
 
 def _get_session(executable: "_ExecutableMixin", session: SessionType = None):
-    from ...deploy.oscar.session import get_default_session
+    from ...session import get_default_session
 
     # if session is not specified, use default session
     if session is None:
@@ -144,7 +144,7 @@ class _ExecutableMixin:
     _executed_sessions: List[SessionType]
 
     def execute(self, session: SessionType = None, **kw):
-        from ...deploy.oscar.session import execute
+        from ...session import execute
 
         session = _get_session(self, session)
         return execute(self, session=session, **kw)
@@ -160,7 +160,7 @@ def _check_session(self, session: SessionType, action: str):
             )
 
     def _fetch(self, session: SessionType = None, **kw):
-        from ...deploy.oscar.session import fetch
+        from ...session import fetch
 
         session = _get_session(self, session)
         self._check_session(session, "fetch")
@@ -175,14 +175,14 @@ def fetch_log(
         offsets: List[int] = None,
         sizes: List[int] = None,
     ):
-        from ...deploy.oscar.session import fetch_log
+        from ...session import fetch_log
 
         session = _get_session(self, session)
         self._check_session(session, "fetch_log")
         return fetch_log(self, session=session, offsets=offsets, sizes=sizes)[0]
 
     def _fetch_infos(self, fields=None, session=None, **kw):
-        from ...deploy.oscar.session import fetch_infos
+        from ...session import fetch_infos
 
         session = _get_session(self, session)
         self._check_session(session, "fetch_infos")
@@ -207,7 +207,7 @@ class _ExecuteAndFetchMixin:
     __slots__ = ()
 
     def _execute_and_fetch(self, session: SessionType = None, **kw):
-        from ...deploy.oscar.session import ExecutionInfo, SyncSession, fetch
+        from ...session import ExecutionInfo, SyncSession, fetch
 
         session = _get_session(self, session)
         fetch_kwargs = kw.pop("fetch_kwargs", dict())
@@ -273,7 +273,7 @@ def __repr__(self):
         return "%s(%s)" % (self._raw_type.__name__, ", ".join(items))
 
     def execute(self, session: SessionType = None, **kw):
-        from ...deploy.oscar.session import execute
+        from ...session import execute
 
         if len(self) == 0:
             return self
@@ -290,14 +290,14 @@ def execute(self, session: SessionType = None, **kw):
             return ret
 
     def _fetch(self, session: SessionType = None, **kw):
-        from ...deploy.oscar.session import fetch
+        from ...session import fetch
 
         session = _get_session(self, session)
         self._check_session(session, "fetch")
         return fetch(*self, session=session, **kw)
 
     def _fetch_infos(self, fields=None, session=None, **kw):
-        from ...deploy.oscar.session import fetch_infos
+        from ...session import fetch_infos
 
         session = _get_session(self, session)
         self._check_session(session, "fetch_infos")
@@ -321,7 +321,7 @@ def fetch_log(
         offsets: List[int] = None,
         sizes: List[int] = None,
     ):
-        from ...deploy.oscar.session import fetch_log
+        from ...session import fetch_log
 
         if len(self) == 0:
             return []

mars/dataframe/base/drop.py

@@ -308,7 +308,7 @@ def df_drop(
 
     Drop columns and/or rows of MultiIndex DataFrame
 
-    >>> midx = pd.MultiIndex(levels=[['lama', 'cow', 'falcon'],
+    >>> midx = pd.MultiIndex(levels=[['lame', 'cow', 'falcon'],
     ...                              ['speed', 'weight', 'length']],
     ...                      codes=[[0, 0, 0, 1, 1, 1, 2, 2, 2],
     ...                             [0, 1, 2, 0, 1, 2, 0, 1, 2]])
@@ -318,7 +318,7 @@ def df_drop(
     ...                         [1, 0.8], [0.3, 0.2]])
     >>> df.execute()
                     big     small
-    lama    speed   45.0    30.0
+    lame    speed   45.0    30.0
             weight  200.0   100.0
             length  1.5     1.0
     cow     speed   30.0    20.0
@@ -330,7 +330,7 @@ def df_drop(
 
     >>> df.drop(index='cow', columns='small').execute()
                     big
-    lama    speed   45.0
+    lame    speed   45.0
             weight  200.0
             length  1.5
     falcon  speed   320.0
@@ -339,7 +339,7 @@ def df_drop(
 
     >>> df.drop(index='length', level=1).execute()
                     big     small
-    lama    speed   45.0    30.0
+    lame    speed   45.0    30.0
             weight  200.0   100.0
     cow     speed   30.0    20.0
             weight  250.0   150.0
@@ -483,14 +483,14 @@ def series_drop(
 
     Drop 2nd level label in MultiIndex Series
 
-    >>> midx = pd.MultiIndex(levels=[['lama', 'cow', 'falcon'],
+    >>> midx = pd.MultiIndex(levels=[['lame', 'cow', 'falcon'],
     ...                              ['speed', 'weight', 'length']],
     ...                      codes=[[0, 0, 0, 1, 1, 1, 2, 2, 2],
     ...                             [0, 1, 2, 0, 1, 2, 0, 1, 2]])
     >>> s = md.Series([45, 200, 1.2, 30, 250, 1.5, 320, 1, 0.3],
     ...               index=midx)
     >>> s.execute()
-    lama    speed      45.0
+    lame    speed      45.0
             weight    200.0
             length      1.2
     cow     speed      30.0
@@ -502,7 +502,7 @@ def series_drop(
     dtype: float64
 
     >>> s.drop(labels='weight', level=1).execute()
-    lama    speed      45.0
+    lame    speed      45.0
             length      1.2
     cow     speed      30.0
             length      1.5

mars/dataframe/base/drop_duplicates.py

@@ -314,14 +314,14 @@ def series_drop_duplicates(series, keep="first", inplace=False, method="auto"):
     Generate a Series with duplicated entries.
 
     >>> import mars.dataframe as md
-    >>> s = md.Series(['lama', 'cow', 'lama', 'beetle', 'lama', 'hippo'],
+    >>> s = md.Series(['lame', 'cow', 'lame', 'beetle', 'lame', 'hippo'],
     ...               name='animal')
     >>> s.execute()
-    0      lama
+    0      lame
     1       cow
-    2      lama
+    2      lame
     3    beetle
-    4      lama
+    4      lame
     5     hippo
     Name: animal, dtype: object
 
@@ -330,7 +330,7 @@ def series_drop_duplicates(series, keep="first", inplace=False, method="auto"):
     set of duplicated entries. The default value of keep is 'first'.
 
     >>> s.drop_duplicates().execute()
-    0      lama
+    0      lame
     1       cow
     3    beetle
     5     hippo
@@ -342,7 +342,7 @@ def series_drop_duplicates(series, keep="first", inplace=False, method="auto"):
     >>> s.drop_duplicates(keep='last').execute()
     1       cow
     3    beetle
-    4      lama
+    4      lame
     5     hippo
     Name: animal, dtype: object
 
@@ -393,20 +393,20 @@ def index_drop_duplicates(index, keep="first", method="auto"):
 
     >>> import mars.dataframe as md
 
-    >>> idx = md.Index(['lama', 'cow', 'lama', 'beetle', 'lama', 'hippo'])
+    >>> idx = md.Index(['lame', 'cow', 'lame', 'beetle', 'lame', 'hippo'])
 
     The `keep` parameter controls  which duplicate values are removed.
     The value 'first' keeps the first occurrence for each
     set of duplicated entries. The default value of keep is 'first'.
 
     >>> idx.drop_duplicates(keep='first').execute()
-    Index(['lama', 'cow', 'beetle', 'hippo'], dtype='object')
+    Index(['lame', 'cow', 'beetle', 'hippo'], dtype='object')
 
     The value 'last' keeps the last occurrence for each set of duplicated
     entries.
 
     >>> idx.drop_duplicates(keep='last').execute()
-    Index(['cow', 'beetle', 'lama', 'hippo'], dtype='object')
+    Index(['cow', 'beetle', 'lame', 'hippo'], dtype='object')
 
     The value ``False`` discards all sets of duplicated entries.
 

mars/dataframe/base/duplicated.py

@@ -427,7 +427,7 @@ def series_duplicated(series, keep="first", method="auto"):
 
     >>> import mars.dataframe as md
 
-    >>> animals = md.Series(['lama', 'cow', 'lama', 'beetle', 'lama'])
+    >>> animals = md.Series(['lame', 'cow', 'lame', 'beetle', 'lame'])
     >>> animals.duplicated().execute()
     0    False
     1    False
@@ -510,7 +510,7 @@ def index_duplicated(index, keep="first"):
 
     >>> import mars.dataframe as md
 
-    >>> idx = md.Index(['lama', 'cow', 'lama', 'beetle', 'lama'])
+    >>> idx = md.Index(['lame', 'cow', 'lame', 'beetle', 'lame'])
     >>> idx.duplicated().execute()
     array([False, False,  True, False,  True])
 

mars/dataframe/base/isin.py

@@ -291,9 +291,9 @@ def series_isin(elements, values):
     Examples
     --------
     >>> import mars.dataframe as md
-    >>> s = md.Series(['lama', 'cow', 'lama', 'beetle', 'lama',
+    >>> s = md.Series(['lame', 'cow', 'lame', 'beetle', 'lame',
     ...                'hippo'], name='animal')
-    >>> s.isin(['cow', 'lama']).execute()
+    >>> s.isin(['cow', 'lame']).execute()
     0     True
     1     True
     2     True
@@ -302,10 +302,10 @@ def series_isin(elements, values):
     5    False
     Name: animal, dtype: bool
 
-    Passing a single string as ``s.isin('lama')`` will raise an error. Use
+    Passing a single string as ``s.isin('lame')`` will raise an error. Use
     a list of one element instead:
 
-    >>> s.isin(['lama']).execute()
+    >>> s.isin(['lame']).execute()
     0     True
     1    False
     2     True

mars/dataframe/contrib/raydataset/tests/test_mldataset.py

@@ -20,7 +20,7 @@
 from ..... import dataframe as md
 from .....conftest import MARS_CI_BACKEND
 from .....deploy.oscar.ray import new_cluster
-from .....deploy.oscar.session import new_session
+from .....session import new_session
 from .....tests.core import require_ray
 from .....utils import lazy_import
 from ....utils import ray_deprecate_ml_dataset

mars/dataframe/contrib/raydataset/tests/test_raydataset.py

@@ -20,7 +20,7 @@
 from ..... import dataframe as md
 from .....conftest import MARS_CI_BACKEND
 from .....deploy.oscar.ray import new_cluster
-from .....deploy.oscar.session import new_session
+from .....session import new_session
 from .....tests.core import require_ray
 from .....utils import lazy_import
 from ....contrib import raydataset as mdd