[DataFrame] Implemented nunique, skew

python/ray/dataframe/dataframe.py

@@ -2672,9 +2672,20 @@ def nsmallest(self, n, columns, keep='first'):
             "github.com/ray-project/ray.")
 
     def nunique(self, axis=0, dropna=True):
-        raise NotImplementedError(
-            "To contribute to Pandas on Ray, please visit "
-            "github.com/ray-project/ray.")
+        """Return Series with number of distinct 
+           observations over requested axis.
+
+        Args:
+            axis : {0 or ‘index’, 1 or ‘columns’}, default 0
+            dropna : boolean, default True
+
+        Returns:
+            nunique : Series
+        """
+        def remote_func(df):
+            return df.nunique(axis=axis, dropna=dropna)
+
+        return self._arithmetic_helper(remote_func, axis)
 
     def pct_change(self, periods=1, fill_method='pad', limit=None, freq=None,
                    **kwargs):
@@ -3408,9 +3419,23 @@ def shift(self, periods=1, freq=None, axis=0):
 
     def skew(self, axis=None, skipna=None, level=None, numeric_only=None,
              **kwargs):
-        raise NotImplementedError(
-            "To contribute to Pandas on Ray, please visit "
-            "github.com/ray-project/ray.")
+        """Return unbiased skew over requested axis Normalized by N-1
+
+        Args:
+            axis : {index (0), columns (1)}
+            skipna : boolean, default True
+            Exclude NA/null values when computing the result.
+            level : int or level name, default None
+            numeric_only : boolean, default None
+
+        Returns:
+            skew : Series or DataFrame (if level specified)
+        """
+        def remote_func(df):
+            return df.skew(axis=axis, skipna=skipna, level=level,
+                          numeric_only=numeric_only, **kwargs)
+
+        return self._arithmetic_helper(remote_func, axis, level)
 
     def slice_shift(self, periods=1, axis=0):
         raise NotImplementedError(

python/ray/dataframe/test/test_dataframe.py

@@ -2149,11 +2149,9 @@ def test_nsmallest():
         ray_df.nsmallest(None, None)
 
 
-def test_nunique():
-    ray_df = create_test_dataframe()
-
-    with pytest.raises(NotImplementedError):
-        ray_df.nunique()
+@pytest.fixture
+def test_nunique(ray_df, pandas_df):
+    assert(ray_df_equals_pandas(ray_df.nunique(), pandas_df.nunique()))
 
 
 def test_pct_change():
@@ -2205,14 +2203,14 @@ def test_pow():
     test_inter_df_math("pow", simple=False)
 
 
-def test_prod():
+def test_prod(ray_df, pandas_df):
     ray_df = create_test_dataframe()
 
     with pytest.raises(NotImplementedError):
-        ray_df.prod(None)
+        ray_df.prod()
 
 
-def test_product():
+def test_product(ray_df, pandas_df):
     ray_df = create_test_dataframe()
 
     with pytest.raises(NotImplementedError):
@@ -2651,11 +2649,9 @@ def test_shift():
         ray_df.shift()
 
 
-def test_skew():
-    ray_df = create_test_dataframe()
-
-    with pytest.raises(NotImplementedError):
-        ray_df.skew()
+@pytest.fixture
+def test_skew(ray_df, pandas_df):
+    assert(ray_df_equals_pandas(ray_df.skew(), pandas_df.skew()))
 
 
 def test_slice_shift():