[DataFrame] Implemented prod, product, added test suite

python/ray/dataframe/dataframe.py

@@ -604,7 +604,8 @@ def groupby(self, by=None, axis=0, level=None, as_index=True, sort=True,
         return DataFrameGroupBy(self, by, axis, level, as_index, sort,
                                 group_keys, squeeze, **kwargs)
 
-    def sum(self, axis=None, skipna=True, level=None, numeric_only=None):
+    def sum(self, axis=None, skipna=True, level=None, numeric_only=None,
+            min_count=1, **kwargs):
         """Perform a sum across the DataFrame.
 
         Args:
@@ -616,7 +617,8 @@ def sum(self, axis=None, skipna=True, level=None, numeric_only=None):
         """
         def remote_func(df):
             return df.sum(axis=axis, skipna=skipna, level=level,
-                          numeric_only=numeric_only)
+                          numeric_only=numeric_only, min_count=min_count,
+                          **kwargs)
 
         return self._arithmetic_helper(remote_func, axis, level)
 
@@ -2740,16 +2742,43 @@ def pow(self, other, axis='columns', level=None, fill_value=None):
                                      fill_value)
 
     def prod(self, axis=None, skipna=None, level=None, numeric_only=None,
-             min_count=0, **kwargs):
-        raise NotImplementedError(
-            "To contribute to Pandas on Ray, please visit "
-            "github.com/ray-project/ray.")
+             min_count=1, **kwargs):
+        """Return the product of the values for the requested axis
+
+        Args:
+            axis : {index (0), columns (1)}
+            skipna : boolean, default True
+            level : int or level name, default None
+            numeric_only : boolean, default None
+            min_count : int, default 1
+
+        Returns:
+            prod : Series or DataFrame (if level specified)
+        """
+        def remote_func(df):
+            return df.prod(axis=axis, skipna=skipna, level=level,
+                           numeric_only=numeric_only, min_count=min_count,
+                           **kwargs)
+
+        return self._arithmetic_helper(remote_func, axis, level)
 
     def product(self, axis=None, skipna=None, level=None, numeric_only=None,
-                min_count=0, **kwargs):
-        raise NotImplementedError(
-            "To contribute to Pandas on Ray, please visit "
-            "github.com/ray-project/ray.")
+                min_count=1, **kwargs):
+        """Return the product of the values for the requested axis
+
+        Args:
+            axis : {index (0), columns (1)}
+            skipna : boolean, default True
+            level : int or level name, default None
+            numeric_only : boolean, default None
+            min_count : int, default 1
+
+        Returns:
+            product : Series or DataFrame (if level specified)
+        """
+        return self.prod(axis=axis, skipna=skipna, level=level,
+                         numeric_only=numeric_only, min_count=min_count,
+                         **kwargs)
 
     def quantile(self, q=0.5, axis=0, numeric_only=True,
                  interpolation='linear'):

python/ray/dataframe/test/test_dataframe.py

@@ -207,6 +207,8 @@ def test_int_dataframe():
 
     test_copy(ray_df)
     test_sum(ray_df, pandas_df)
+    test_prod(ray_df, pandas_df)
+    test_product(ray_df, pandas_df)
     test_abs(ray_df, pandas_df)
     test_keys(ray_df, pandas_df)
     test_transpose(ray_df, pandas_df)
@@ -367,6 +369,8 @@ def test_float_dataframe():
 
     test_copy(ray_df)
     test_sum(ray_df, pandas_df)
+    test_prod(ray_df, pandas_df)
+    test_product(ray_df, pandas_df)
     test_abs(ray_df, pandas_df)
     test_keys(ray_df, pandas_df)
     test_transpose(ray_df, pandas_df)
@@ -2205,18 +2209,14 @@ def test_pow():
     test_inter_df_math("pow", simple=False)
 
 
-def test_prod():
-    ray_df = create_test_dataframe()
-
-    with pytest.raises(NotImplementedError):
-        ray_df.prod(None)
-
+@pytest.fixture
+def test_prod(ray_df, pandas_df):
+    assert(ray_df.prod().equals(pandas_df.prod()))
 
-def test_product():
-    ray_df = create_test_dataframe()
 
-    with pytest.raises(NotImplementedError):
-        ray_df.product()
+@pytest.fixture
+def test_product(ray_df, pandas_df):
+    assert(ray_df.product().equals(pandas_df.product()))
 
 
 @pytest.fixture