This is an automated email from the ASF dual-hosted git repository.
ruifengz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 398bff77b1c [SPARK-45999][PS] Use dedicated `PandasProduct` in
`cumprod`
398bff77b1c is described below
commit 398bff77b1c837aed55f4f10ff1157f7da813570
Author: Ruifeng Zheng <ruife...@apache.org>
AuthorDate: Mon Nov 20 16:55:09 2023 +0800
[SPARK-45999][PS] Use dedicated `PandasProduct` in `cumprod`
### What changes were proposed in this pull request?
Use dedicated `PandasProduct` in `cumprod`
### Why are the changes needed?
to be consistent with the `prod`
### Does this PR introduce _any_ user-facing change?
no
### How was this patch tested?
ci
### Was this patch authored or co-authored using generative AI tooling?
no
Closes #43901 from zhengruifeng/ps_cumprod.
Authored-by: Ruifeng Zheng <ruife...@apache.org>
Signed-off-by: Ruifeng Zheng <ruife...@apache.org>
---
python/pyspark/pandas/generic.py | 2 --
python/pyspark/pandas/series.py | 34 +++++++---------------------------
2 files changed, 7 insertions(+), 29 deletions(-)
diff --git a/python/pyspark/pandas/generic.py b/python/pyspark/pandas/generic.py
index 23139762882..77cefb53fe5 100644
--- a/python/pyspark/pandas/generic.py
+++ b/python/pyspark/pandas/generic.py
@@ -331,8 +331,6 @@ class Frame(object, metaclass=ABCMeta):
return self._apply_series_op(lambda psser: psser._cumsum(skipna),
should_resolve=True)
# TODO: add 'axis' parameter
- # TODO: use pandas_udf to support negative values and other options later
- # other window except unbounded ones is supported as of Spark 3.0.
def cumprod(self: FrameLike, skipna: bool = True) -> FrameLike:
"""
Return cumulative product over a DataFrame or Series axis.
diff --git a/python/pyspark/pandas/series.py b/python/pyspark/pandas/series.py
index c9beb6432f9..fedc2417a29 100644
--- a/python/pyspark/pandas/series.py
+++ b/python/pyspark/pandas/series.py
@@ -62,7 +62,6 @@ from pyspark.sql.types import (
DoubleType,
FloatType,
IntegerType,
- IntegralType,
LongType,
NumericType,
Row,
@@ -6973,36 +6972,17 @@ class Series(Frame, IndexOpsMixin, Generic[T]):
return psser._cum(F.sum, skipna, part_cols)
def _cumprod(self, skipna: bool, part_cols: Sequence["ColumnOrName"] = ())
-> "Series":
- if isinstance(self.spark.data_type, BooleanType):
- scol = self._cum(
- lambda scol: F.min(F.coalesce(scol, F.lit(True))), skipna,
part_cols
- ).spark.column.cast(LongType())
- elif isinstance(self.spark.data_type, NumericType):
- num_zeros = self._cum(
- lambda scol: F.sum(F.when(scol == 0, 1).otherwise(0)), skipna,
part_cols
- ).spark.column
- num_negatives = self._cum(
- lambda scol: F.sum(F.when(scol < 0, 1).otherwise(0)), skipna,
part_cols
- ).spark.column
- sign = F.when(num_negatives % 2 == 0, 1).otherwise(-1)
-
- abs_prod = F.exp(
- self._cum(lambda scol: F.sum(F.log(F.abs(scol))), skipna,
part_cols).spark.column
- )
-
- scol = F.when(num_zeros > 0, 0).otherwise(sign * abs_prod)
-
- if isinstance(self.spark.data_type, IntegralType):
- scol = F.round(scol).cast(LongType())
- else:
+ psser = self
+ if isinstance(psser.spark.data_type, BooleanType):
+ psser = psser.spark.transform(lambda scol: scol.cast(LongType()))
+ elif not isinstance(psser.spark.data_type, NumericType):
raise TypeError(
"Could not convert {} ({}) to numeric".format(
- spark_type_to_pandas_dtype(self.spark.data_type),
- self.spark.data_type.simpleString(),
+ spark_type_to_pandas_dtype(psser.spark.data_type),
+ psser.spark.data_type.simpleString(),
)
)
-
- return self._with_new_scol(scol)
+ return psser._cum(lambda c: SF.product(c, skipna), skipna, part_cols)
# ----------------------------------------------------------------------
# Accessor Methods
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
