[BACKPORT] Fix accuracy_score for distributed execution (#1945) (#1948)

Co-authored-by: wenjun.swj <[email protected]>

Co-authored-by: Xuye (Chris) Qin <[email protected]>

Author: wjsi

mars/dataframe/core.py

@@ -512,6 +512,14 @@ def __str__(self):
     def __repr__(self):
         return self._to_str(representation=True)
 
+    def _to_mars_tensor(self, dtype=None, order='K', extract_multi_index=False):
+        tensor = self.to_tensor(extract_multi_index=extract_multi_index)
+        dtype = dtype if dtype is not None else tensor.dtype
+        return tensor.astype(dtype=dtype, order=order, copy=False)
+
+    def __mars_tensor__(self, dtype=None, order='K'):
+        return self._to_mars_tensor(dtype=dtype, order=order)
+
     @property
     def dtype(self):
         return getattr(self, '_dtype', None) or self.op.dtype
@@ -553,13 +561,8 @@ def __new__(cls, data: Union[pd.Index, IndexData], **_):
     def __len__(self):
         return len(self._data)
 
-    def _to_mars_tensor(self, dtype=None, order='K', extract_multi_index=False):
-        tensor = self._data.to_tensor(extract_multi_index=extract_multi_index)
-        dtype = dtype if dtype is not None else tensor.dtype
-        return tensor.astype(dtype=dtype, order=order, copy=False)
-
     def __mars_tensor__(self, dtype=None, order='K'):
-        return self._to_mars_tensor(dtype=dtype, order=order)
+        return self._data.__mars_tensor__(dtype=dtype, order=order)
 
     def _get_df_or_series(self):
         obj = getattr(self, '_df_or_series', None)
@@ -597,6 +600,10 @@ def names(self, value):
         else:
             self.rename(value, inplace=True)
 
+    @property
+    def values(self):
+        return self.to_tensor()
+
     def to_frame(self, index: bool = True, name=None):
         """
         Create a DataFrame with a column containing the Index.
@@ -663,7 +670,7 @@ def to_frame(self, index: bool = True, name=None):
         else:
             columns = [name or self.name or 0]
         index_ = self if index else None
-        return dataframe_from_tensor(self._to_mars_tensor(self, extract_multi_index=True),
+        return dataframe_from_tensor(self._data._to_mars_tensor(self, extract_multi_index=True),
                                      index=index_, columns=columns)
 
     def to_series(self, index=None, name=None):
@@ -889,6 +896,11 @@ def from_tensor(in_tensor, index=None, name=None):
 class SeriesData(_BatchedFetcher, BaseSeriesData):
     _type_name = 'Series'
 
+    def __mars_tensor__(self, dtype=None, order='K'):
+        tensor = self.to_tensor()
+        dtype = dtype if dtype is not None else tensor.dtype
+        return tensor.astype(dtype=dtype, order=order, copy=False)
+
     @classmethod
     def cls(cls, provider):
         if provider.type == ProviderType.protobuf:
@@ -1002,9 +1014,22 @@ def __len__(self):
         return len(self._data)
 
     def __mars_tensor__(self, dtype=None, order='K'):
-        tensor = self._data.to_tensor()
-        dtype = dtype if dtype is not None else tensor.dtype
-        return tensor.astype(dtype=dtype, order=order, copy=False)
+        return self._data.__mars_tensor__(dtype=dtype, order=order)
+
+    def keys(self):
+        """
+        Return alias for index.
+
+        Returns
+        -------
+        Index
+            Index of the Series.
+        """
+        return self.index
+
+    @property
+    def values(self):
+        return self.to_tensor()
 
     def iteritems(self, batch_size=10000, session=None):
         """
@@ -1288,6 +1313,9 @@ def __str__(self):
     def __repr__(self):
         return self._to_str(representation=True)
 
+    def __mars_tensor__(self, dtype=None, order='K'):
+        return self.to_tensor().astype(dtype=dtype, order=order, copy=False)
+
     def _repr_html_(self):
         if len(self._executed_sessions) == 0:
             # not executed before, fall back to normal repr
@@ -1350,7 +1378,7 @@ def from_records(self, records, **kw):
         return self._data.from_records(records, **kw)
 
     def __mars_tensor__(self, dtype=None, order='K'):
-        return self._data.to_tensor().astype(dtype=dtype, order=order, copy=False)
+        return self._data.__mars_tensor__(dtype=dtype, order=order)
 
     def __getattr__(self, key):
         try:
@@ -1409,6 +1437,23 @@ def columns(self, new_columns):
         new_df = op(self)
         self.data = new_df.data
 
+    def keys(self):
+        """
+        Get the 'info axis' (see Indexing for more).
+
+        This is index for Series, columns for DataFrame.
+
+        Returns
+        -------
+        Index
+            Info axis.
+        """
+        return self.columns
+
+    @property
+    def values(self):
+        return self.to_tensor()
+
     @property
     def dtypes(self):
         """

mars/dataframe/tests/test_core.py

@@ -94,3 +94,28 @@ def testToFrameOrSeries(self):
         r = index.to_series(name='new_name')
         result = self.executor.execute_dataframe(r, concat=True)[0]
         pd.testing.assert_series_equal(raw.to_series(name='new_name'), result)
+
+    def testKeyValue(self):
+        raw = pd.DataFrame(np.random.rand(4, 3), columns=list('ABC'))
+        df = DataFrame(raw)
+
+        result = self.executor.execute_dataframe(df.values, concat=True)[0]
+        np.testing.assert_array_equal(result, raw.values)
+
+        result = self.executor.execute_dataframe(df.keys(), concat=True)[0]
+        pd.testing.assert_index_equal(result, raw.keys())
+
+        raw = pd.Series(np.random.rand(10))
+        s = Series(raw)
+
+        result = self.executor.execute_dataframe(s.values, concat=True)[0]
+        np.testing.assert_array_equal(result, raw.values)
+
+        result = self.executor.execute_dataframe(s.keys(), concat=True)[0]
+        pd.testing.assert_index_equal(result, raw.keys())
+
+        raw = pd.Index(np.random.rand(10))
+        idx = Index(raw)
+
+        result = self.executor.execute_dataframe(idx.values, concat=True)[0]
+        np.testing.assert_array_equal(result, raw.values)

mars/learn/metrics/_check_targets.py

@@ -16,7 +16,7 @@
 
 from ... import opcodes as OperandDef
 from ... import tensor as mt
-from ...core import Base, Entity, ExecutableTuple, get_output_types
+from ...core import Base, Entity, ExecutableTuple
 from ...context import get_context
 from ...serialize import AnyField, KeyField
 from ...tiles import TilesError
@@ -39,9 +39,7 @@ def __init__(self, y_true=None, y_pred=None, type_true=None, type_pred=None, **k
         super().__init__(_y_true=y_true, _y_pred=y_pred,
                          _type_true=type_true, _type_pred=type_pred, **kw)
         # scalar(y_type), y_true, y_pred
-        self.output_types = \
-            [OutputType.tensor] + get_output_types(*[y_true, y_pred],
-                                                   unknown_as=OutputType.tensor)
+        self.output_types = [OutputType.tensor] * 3
 
     @property
     def output_limit(self):
@@ -111,7 +109,7 @@ def tile(cls, op):
             type_true, type_pred = ctx.get_chunk_results(
                 [op.type_true.chunks[0].key,
                  op.type_pred.chunks[0].key])
-        except KeyError:
+        except (KeyError, AttributeError):
             raise TilesError('type_true and type_pred '
                              'needs to be executed first')
 

mars/learn/metrics/_classification.py

@@ -90,7 +90,7 @@ def tile(cls, op):
         ctx = get_context()
         try:
             type_true = ctx.get_chunk_results([op.type_true.chunks[0].key])[0]
-        except KeyError:
+        except (KeyError, AttributeError):
             raise TilesError('type_true needed to be executed first')
 
         y_true, y_pred = op.y_true, op.y_pred

mars/learn/metrics/tests/integrated/test_ranking.py

@@ -20,12 +20,13 @@
 import pandas as pd
 try:
     import sklearn
-    from sklearn.metrics import roc_curve as sklearn_roc_curve, auc as sklearn_auc
+    from sklearn.metrics import roc_curve as sklearn_roc_curve, auc as sklearn_auc, \
+        accuracy_score as sklearn_accuracy_score
 except ImportError:
     sklearn = None
 
 from mars import dataframe as md
-from mars.learn.metrics import roc_curve, auc
+from mars.learn.metrics import roc_curve, auc, accuracy_score
 from mars.tests.integrated.base import IntegrationTestBase
 from mars.session import new_session
 
@@ -55,3 +56,22 @@ def testRocCurveAuc(self):
                                                             pos_label=2)
             expect_m = sklearn_auc(sk_fpr, sk_tpr)
             self.assertAlmostEqual(m.fetch(session=sess), expect_m)
+
+    def testAccuracyScore(self):
+        service_ep = 'http://127.0.0.1:' + self.web_port
+        timeout = 120 if 'CI' in os.environ else -1
+        with new_session(service_ep) as sess:
+            run_kwargs = {'timeout': timeout}
+
+            rs = np.random.RandomState(0)
+            raw = pd.DataFrame({'a': rs.randint(0, 10, (10,)),
+                                'b': rs.randint(0, 10, (10,))})
+
+            df = md.DataFrame(raw)
+            y = df['a'].to_tensor().astype('int')
+            pred = df['b'].astype('int')
+
+            score = accuracy_score(y, pred, session=sess, run_kwargs=run_kwargs)
+            expect = sklearn_accuracy_score(raw['a'].to_numpy().astype('int'),
+                                            raw['b'].to_numpy().astype('int'))
+            self.assertAlmostEqual(score.fetch(session=sess), expect)

mars/learn/utils/multiclass.py

@@ -398,7 +398,8 @@ def type_of_target(y):
     'multilabel-indicator'
     """
     valid_types = (Sequence, spmatrix) if spmatrix is not None else (Sequence,)
-    valid = ((isinstance(y, valid_types) or hasattr(y, '__array__'))
+    valid = ((isinstance(y, valid_types) or
+              hasattr(y, '__array__') or hasattr(y, '__mars_tensor__'))
              and not isinstance(y, str))
 
     if not valid: