You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/04/10 04:12:49 UTC
[GitHub] [spark] zhengruifeng opened a new pull request, #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
zhengruifeng opened a new pull request, #36127:
URL: https://github.com/apache/spark/pull/36127
### What changes were proposed in this pull request?
Implement linear interpolate, see [SPARK-38844](https://issues.apache.org/jira/browse/SPARK-38844)
### Why are the changes needed?
Increase pandas API coverage in PySpark
### Does this PR introduce _any_ user-facing change?
yes, new function was added:
```
In [3]: df = ps.DataFrame({'s1': [-1, .2, np.nan, np.nan, np.nan, .6, .2, .4, .5, .6, np.nan, np.nan], 's2': [np.nan, 2, np.nan, 1, 3, 1, np.nan, np.nan, 0, 0, np.nan, 2]})
In [4]: df.interpolate(method='linear')
s1 s2
0 -1.0 NaN
1 0.2 2.000000
2 0.3 1.500000
3 0.4 1.000000
4 0.5 3.000000
5 0.6 1.000000
6 0.2 0.666667
7 0.4 0.333333
8 0.5 0.000000
9 0.6 0.000000
10 0.6 1.000000
11 0.6 2.000000
```
### How was this patch tested?
added UT
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r853600923
##########
python/pyspark/pandas/tests/test_generic_functions.py:
##########
@@ -0,0 +1,124 @@
+#
Review Comment:
https://github.com/apache/spark/pull/36271
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r846917155
##########
python/pyspark/pandas/frame.py:
##########
@@ -5500,6 +5500,25 @@ def op(psser: ps.Series) -> ps.Series:
else:
return psdf
+ def interpolate(
+ self, method: Optional[str] = None, limit: Optional[int] = None
+ ) -> Optional["DataFrame"]:
Review Comment:
I think it always returns a `DataFrame`. Maybe we should change `Optional["DataFrame"]`to `"DataFrame"`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on PR #36127:
URL: https://github.com/apache/spark/pull/36127#issuecomment-1094249727
cc @HyukjinKwon
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r846935976
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/windowExpressions.scala:
##########
@@ -1014,3 +1014,77 @@ case class PercentRank(children: Seq[Expression]) extends RankLike with SizeBase
override protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): PercentRank =
copy(children = newChildren)
}
+
+/**
+ * Keep the Last non-Null value if any.
+ * MUST only be used in the Pandas API on Pyspark.
+ * input seq: null, 1, 2, 3, null, 4, 5, null
+ * output seq: null, 1, 2, 3, 3, 4, 5, 5
+ */
+private[sql] case class LastNonNull(input: Expression)
+ extends AggregateWindowFunction with UnaryLike[Expression] {
+
+ override def dataType: DataType = input.dataType
+
+ private val nan = Literal.create(null, dataType)
Review Comment:
```suggestion
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/windowExpressions.scala:
##########
@@ -1014,3 +1014,77 @@ case class PercentRank(children: Seq[Expression]) extends RankLike with SizeBase
override protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): PercentRank =
copy(children = newChildren)
}
+
+/**
+ * Keep the Last non-Null value if any.
+ * MUST only be used in the Pandas API on Pyspark.
+ * input seq: null, 1, 2, 3, null, 4, 5, null
+ * output seq: null, 1, 2, 3, 3, 4, 5, 5
+ */
+private[sql] case class LastNonNull(input: Expression)
+ extends AggregateWindowFunction with UnaryLike[Expression] {
+
+ override def dataType: DataType = input.dataType
+
+ private val nan = Literal.create(null, dataType)
+ private val last = AttributeReference("last", dataType, nullable = true)()
+
+ override def aggBufferAttributes: Seq[AttributeReference] = last :: Nil
+
+ override val initialValues: Seq[Expression] = Seq(nan)
Review Comment:
```suggestion
override val initialValues: Seq[Expression] = Seq(Literal.create(null, dataType))
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on PR #36127:
URL: https://github.com/apache/spark/pull/36127#issuecomment-1097524188
Thanks all!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on PR #36127:
URL: https://github.com/apache/spark/pull/36127#issuecomment-1097775178
Build link: https://github.com/zhengruifeng/spark/actions/runs/2160215300
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r846925408
##########
python/pyspark/pandas/series.py:
##########
@@ -2168,6 +2169,66 @@ def _fillna(
)
)._psser_for(self._column_label)
+ def interpolate(
+ self, method: Optional[str] = None, limit: Optional[int] = None
+ ) -> Optional["Series"]:
+ return self._interpolate(method=method, limit=limit)
+
+ def _interpolate(
+ self,
+ method: Optional[str] = None,
+ limit: Optional[int] = None,
+ ) -> "Series":
+ if (method is not None) and (method not in ["linear"]):
+ raise ValueError("Expecting 'linear'.")
+ if (limit is not None) and (not limit > 0):
+ raise ValueError("limit must be > 0.")
+
+ if not self.spark.nullable and not isinstance(
+ self.spark.data_type, (FloatType, DoubleType)
+ ):
+ return self._psdf.copy()._psser_for(self._column_label)
+
+ scol = self.spark.column
+ sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
+ last_non_null = Column(sql_utils.lastNonNull(scol._jc))
+ null_index = Column(sql_utils.nullIndex(scol._jc))
+
+ window_forward = Window.orderBy(NATURAL_ORDER_COLUMN_NAME).rowsBetween(
+ Window.unboundedPreceding, Window.currentRow
+ )
+ last_non_null_forward = last_non_null.over(window_forward)
+ null_index_forward = null_index.over(window_forward)
+
+ window_backward = Window.orderBy(F.desc(NATURAL_ORDER_COLUMN_NAME)).rowsBetween(
+ Window.unboundedPreceding, Window.currentRow
+ )
+ last_non_null_backward = last_non_null.over(window_backward)
+ null_index_backward = null_index.over(window_backward)
+
+ fill = (last_non_null_backward - last_non_null_forward) / (
+ null_index_backward + null_index_forward
+ ) * null_index_forward + last_non_null_forward
+
+ fill_cond = ~F.isnull(last_non_null_backward) & ~F.isnull(last_non_null_forward)
+ pad_cond = F.isnull(last_non_null_backward) & ~F.isnull(last_non_null_forward)
+ if limit is not None:
+ fill_cond = fill_cond & (null_index_forward <= F.lit(limit))
+ pad_cond = pad_cond & (null_index_forward <= F.lit(limit))
+
+ cond = self.isnull().spark.column
+ scol = (
+ F.when(cond & fill_cond, fill)
+ .when(cond & pad_cond, last_non_null_forward)
+ .otherwise(scol)
+ )
+
+ return DataFrame(
+ self._psdf._internal.with_new_spark_column(
+ self._column_label, scol.alias(name_like_string(self.name)) # TODO: dtype?
Review Comment:
I think we don't have to bother with aliasing here.`scol.alias(name_like_string(self.name)` -> `scol`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] itholic commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
itholic commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r846943416
##########
python/pyspark/pandas/generic.py:
##########
@@ -3181,6 +3181,83 @@ def ffill(
pad = ffill
+ # TODO: add 'axis', 'inplace', 'limit_direction', 'limit_area', 'downcast'
+ def interpolate(
+ self: FrameLike,
+ method: Optional[str] = None,
+ limit: Optional[int] = None,
+ ) -> FrameLike:
+ """
+ Fill NaN values using an interpolation method.
+
+ Parameters
+ ----------
+ method : str, default 'linear'
+ Interpolation technique to use. One of:
+
+ * 'linear': Ignore the index and treat the values as equally
+ spaced.
+
+ limit : int, optional
+ Maximum number of consecutive NaNs to fill. Must be greater than
+ 0.
+
+ Returns
+ -------
+ Series or DataFrame or None
+ Returns the same object type as the caller, interpolated at
+ some or all NA values.
+
+ See Also
+ --------
+ fillna : Fill missing values using different methods.
+
+ Examples
+ --------
+ Filling in NA via linear interpolation.
+
+ >>> s = ps.Series([0, 1, np.nan, 3])
+ >>> s
+ 0 0.0
+ 1 1.0
+ 2 NaN
+ 3 3.0
+ dtype: float64
+ >>> s.interpolate()
+ 0 0.0
+ 1 1.0
+ 2 2.0
+ 3 3.0
+ dtype: float64
+
+ Fill the DataFrame forward (that is, going down) along each column
+ using linear interpolation.
+
+ Note how the last entry in column 'a' is interpolated differently,
+ because there is no entry after it to use for interpolation.
+ Note how the first entry in column 'b' remains NA, because there
+ is no entry before it to use for interpolation.
+
+ >>> df = ps.DataFrame([(0.0, np.nan, -1.0, 1.0),
+ ... (np.nan, 2.0, np.nan, np.nan),
+ ... (2.0, 3.0, np.nan, 9.0),
+ ... (np.nan, 4.0, -4.0, 16.0)],
+ ... columns=list('abcd'))
+ >>> df
+ a b c d
+ 0 0.0 NaN -1.0 1.0
+ 1 NaN 2.0 NaN NaN
+ 2 2.0 3.0 NaN 9.0
+ 3 NaN 4.0 -4.0 16.0
+ >>> df.interpolate(method='linear')
+ a b c d
+ 0 0.0 NaN -1.0 1.0
+ 1 1.0 2.0 -2.0 5.0
+ 2 2.0 3.0 -3.0 9.0
+ 3 2.0 4.0 -4.0 16.0
Review Comment:
You could refer to:
https://github.com/apache/spark/blob/2f122ba6d13ea26411fa4bf3e636ced449a8a205/python/pyspark/pandas/frame.py#L4032-L4035
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #36127:
URL: https://github.com/apache/spark/pull/36127#issuecomment-1097685882
@zhengruifeng mind rebasing please? Otherwise should be good to go
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r846984886
##########
python/pyspark/pandas/frame.py:
##########
@@ -5500,6 +5500,25 @@ def op(psser: ps.Series) -> ps.Series:
else:
return psdf
+ def interpolate(
+ self, method: Optional[str] = None, limit: Optional[int] = None
+ ) -> Optional["DataFrame"]:
Review Comment:
Yes. It should be a `"DataFrame"` before we add param `inplace`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r853597396
##########
python/pyspark/pandas/tests/test_generic_functions.py:
##########
@@ -0,0 +1,124 @@
+#
Review Comment:
Oh, actually we should add this file into https://github.com/apache/spark/blob/master/dev/sparktestsupport/modules.py#L440
##########
python/pyspark/pandas/tests/test_generic_functions.py:
##########
@@ -0,0 +1,124 @@
+#
Review Comment:
let me make a quick followup
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r846919520
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/windowExpressions.scala:
##########
@@ -1014,3 +1014,77 @@ case class PercentRank(children: Seq[Expression]) extends RankLike with SizeBase
override protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): PercentRank =
copy(children = newChildren)
}
+
+/**
+ * Keep the Last non-Null value if any.
+ * MUST only be used in the Pandas API on Pyspark.
+ * input seq: null, 1, 2, 3, null, 4, 5, null
+ * output seq: null, 1, 2, 3, 3, 4, 5, 5
+ */
+private[sql] case class LastNonNull(input: Expression)
Review Comment:
`private[sql]` can be all removed per SPARK-16813
```suggestion
case class LastNonNull(input: Expression)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r846935870
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/windowExpressions.scala:
##########
@@ -1014,3 +1014,77 @@ case class PercentRank(children: Seq[Expression]) extends RankLike with SizeBase
override protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): PercentRank =
copy(children = newChildren)
}
+
+/**
+ * Keep the Last non-Null value if any.
+ * MUST only be used in the Pandas API on Pyspark.
+ * input seq: null, 1, 2, 3, null, 4, 5, null
+ * output seq: null, 1, 2, 3, 3, 4, 5, 5
Review Comment:
```suggestion
* Keep the last non-null value seen if any. This expression is dedicated only for Pandas API on Spark.
* For example,
* Input: null, 1, 2, 3, null, 4, 5, null
* Output: null, 1, 2, 3, 3, 4, 5, 5
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r846918678
##########
python/pyspark/pandas/generic.py:
##########
@@ -3181,6 +3181,83 @@ def ffill(
pad = ffill
+ # TODO: add 'axis', 'inplace', 'limit_direction', 'limit_area', 'downcast'
+ def interpolate(
+ self: FrameLike,
+ method: Optional[str] = None,
+ limit: Optional[int] = None,
+ ) -> FrameLike:
+ """
+ Fill NaN values using an interpolation method.
+
+ Parameters
+ ----------
+ method : str, default 'linear'
+ Interpolation technique to use. One of:
+
+ * 'linear': Ignore the index and treat the values as equally
+ spaced.
+
+ limit : int, optional
+ Maximum number of consecutive NaNs to fill. Must be greater than
+ 0.
+
Review Comment:
Maybe we should start adding `versionadded: 3.4.0`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r846918461
##########
python/pyspark/pandas/frame.py:
##########
@@ -5500,6 +5500,25 @@ def op(psser: ps.Series) -> ps.Series:
else:
return psdf
+ def interpolate(
Review Comment:
It should be listed in https://github.com/apache/spark/blob/master/python/docs/source/reference/pyspark.pandas/frame.rst to document.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r847002450
##########
python/pyspark/pandas/series.py:
##########
@@ -2168,6 +2169,66 @@ def _fillna(
)
)._psser_for(self._column_label)
+ def interpolate(
+ self, method: Optional[str] = None, limit: Optional[int] = None
+ ) -> Optional["Series"]:
+ return self._interpolate(method=method, limit=limit)
+
+ def _interpolate(
+ self,
+ method: Optional[str] = None,
+ limit: Optional[int] = None,
+ ) -> "Series":
+ if (method is not None) and (method not in ["linear"]):
+ raise ValueError("Expecting 'linear'.")
+ if (limit is not None) and (not limit > 0):
+ raise ValueError("limit must be > 0.")
+
+ if not self.spark.nullable and not isinstance(
+ self.spark.data_type, (FloatType, DoubleType)
+ ):
+ return self._psdf.copy()._psser_for(self._column_label)
+
+ scol = self.spark.column
+ sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
+ last_non_null = Column(sql_utils.lastNonNull(scol._jc))
+ null_index = Column(sql_utils.nullIndex(scol._jc))
+
+ window_forward = Window.orderBy(NATURAL_ORDER_COLUMN_NAME).rowsBetween(
Review Comment:
Based on SQL, I had no idea either.
but actually, this computation can be further optimzed in a scaled way:
1, perform global sort by `NATURAL_ORDER_COLUMN_NAME`;
2, extra pass on the sorted dataframe: collect some statistics within each physical partition, in this case, it needs: `size of the first and last null sequence`, `the first and last non-null value`.
3, based on the statistics collected from each physical partition, we then compute the state at the begining of each partition, and so can eval the expressions on each rows.
This computation was quite similar to what the `roc` is computed in `BinaryClassificationMetrics`, see https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/mllib/evaluation/BinaryClassificationMetrics.scala#L178-L253 (1, `sortByKey(ascending = false)`; 2, compute the `partitionwiseCumulativeCounts`, 3, fetch the `cumCount` on each partition and compute the curve).
When implementing `ewm`, I also tried this method (see https://github.com/apache/spark/pull/36062/files).
Comparing with exisiting plan, it should be more scalable, but if the dataset is small enough, it maybe solwer since it requires an extra pass.
I think it maybe up to the SQL optimizer to support computing the `window expression with no partition` in a more scaled way, since it seems a generic pattern (`row_number`/`rank`/`dense_rank` in SQL, `autocorr`/`ewm`/`null_index`/`last_non_null` in Python, can all be computed in this way.).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #36127:
URL: https://github.com/apache/spark/pull/36127#issuecomment-1097931058
Merged to master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon closed pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon closed pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
URL: https://github.com/apache/spark/pull/36127
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #36127:
URL: https://github.com/apache/spark/pull/36127#issuecomment-1094539526
Looks pretty good otherwise. Thanks for working on this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #36127:
URL: https://github.com/apache/spark/pull/36127#issuecomment-1094452207
cc @ueshin @itholic @xinrong-databricks too
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on PR #36127:
URL: https://github.com/apache/spark/pull/36127#issuecomment-1094625608
Thanks for reviewing, will update according to your comments!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r846923142
##########
python/pyspark/pandas/generic.py:
##########
@@ -3181,6 +3181,83 @@ def ffill(
pad = ffill
+ # TODO: add 'axis', 'inplace', 'limit_direction', 'limit_area', 'downcast'
+ def interpolate(
+ self: FrameLike,
+ method: Optional[str] = None,
+ limit: Optional[int] = None,
+ ) -> FrameLike:
+ """
+ Fill NaN values using an interpolation method.
+
+ Parameters
+ ----------
+ method : str, default 'linear'
+ Interpolation technique to use. One of:
+
+ * 'linear': Ignore the index and treat the values as equally
+ spaced.
+
+ limit : int, optional
+ Maximum number of consecutive NaNs to fill. Must be greater than
+ 0.
+
+ Returns
+ -------
+ Series or DataFrame or None
+ Returns the same object type as the caller, interpolated at
+ some or all NA values.
+
+ See Also
+ --------
+ fillna : Fill missing values using different methods.
+
+ Examples
+ --------
+ Filling in NA via linear interpolation.
+
+ >>> s = ps.Series([0, 1, np.nan, 3])
+ >>> s
+ 0 0.0
+ 1 1.0
+ 2 NaN
+ 3 3.0
+ dtype: float64
+ >>> s.interpolate()
+ 0 0.0
+ 1 1.0
+ 2 2.0
+ 3 3.0
+ dtype: float64
+
+ Fill the DataFrame forward (that is, going down) along each column
+ using linear interpolation.
+
+ Note how the last entry in column 'a' is interpolated differently,
+ because there is no entry after it to use for interpolation.
+ Note how the first entry in column 'b' remains NA, because there
+ is no entry before it to use for interpolation.
+
+ >>> df = ps.DataFrame([(0.0, np.nan, -1.0, 1.0),
+ ... (np.nan, 2.0, np.nan, np.nan),
+ ... (2.0, 3.0, np.nan, 9.0),
+ ... (np.nan, 4.0, -4.0, 16.0)],
+ ... columns=list('abcd'))
+ >>> df
+ a b c d
+ 0 0.0 NaN -1.0 1.0
+ 1 NaN 2.0 NaN NaN
+ 2 2.0 3.0 NaN 9.0
+ 3 NaN 4.0 -4.0 16.0
+ >>> df.interpolate(method='linear')
+ a b c d
+ 0 0.0 NaN -1.0 1.0
+ 1 1.0 2.0 -2.0 5.0
+ 2 2.0 3.0 -3.0 9.0
+ 3 2.0 4.0 -4.0 16.0
Review Comment:
I think we should better probably add `Notes` section with describing that this API is expensive because Window functions will be executed within one executor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r846921419
##########
python/pyspark/pandas/series.py:
##########
@@ -2168,6 +2169,66 @@ def _fillna(
)
)._psser_for(self._column_label)
+ def interpolate(
+ self, method: Optional[str] = None, limit: Optional[int] = None
+ ) -> Optional["Series"]:
+ return self._interpolate(method=method, limit=limit)
+
+ def _interpolate(
+ self,
+ method: Optional[str] = None,
+ limit: Optional[int] = None,
+ ) -> "Series":
+ if (method is not None) and (method not in ["linear"]):
+ raise ValueError("Expecting 'linear'.")
+ if (limit is not None) and (not limit > 0):
+ raise ValueError("limit must be > 0.")
+
+ if not self.spark.nullable and not isinstance(
+ self.spark.data_type, (FloatType, DoubleType)
+ ):
+ return self._psdf.copy()._psser_for(self._column_label)
+
+ scol = self.spark.column
+ sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
+ last_non_null = Column(sql_utils.lastNonNull(scol._jc))
+ null_index = Column(sql_utils.nullIndex(scol._jc))
+
+ window_forward = Window.orderBy(NATURAL_ORDER_COLUMN_NAME).rowsBetween(
Review Comment:
Oh, okay, I got why this is tricky now. It should know the previous value to compute the current value .. but this way will move all data into a single worker as partition specification cannot be provided. Hm, it will be expensive but I don't have other ideas either.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on PR #36127:
URL: https://github.com/apache/spark/pull/36127#issuecomment-1097687309
@HyukjinKwon Ok, I'm going to rebase it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r847901607
##########
python/pyspark/pandas/frame.py:
##########
@@ -5500,6 +5500,25 @@ def op(psser: ps.Series) -> ps.Series:
else:
return psdf
+ def interpolate(
+ self, method: Optional[str] = None, limit: Optional[int] = None
+ ) -> Optional["DataFrame"]:
+ if (method is not None) and (method not in ["linear"]):
+ raise ValueError("Expecting 'linear'.")
Review Comment:
good catch! will update
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on PR #36127:
URL: https://github.com/apache/spark/pull/36127#issuecomment-1097707147
Build link: https://github.com/zhengruifeng/spark/actions/runs/2159967718
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on PR #36127:
URL: https://github.com/apache/spark/pull/36127#issuecomment-1097932192
Thanks all for reviewing!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
zhengruifeng commented on PR #36127:
URL: https://github.com/apache/spark/pull/36127#issuecomment-1097927622
all tests passed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r846920098
##########
python/pyspark/pandas/tests/test_interpolate.py:
##########
@@ -0,0 +1,122 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import numpy as np
+import pandas as pd
+
+import pyspark.pandas as ps
+from pyspark.testing.pandasutils import PandasOnSparkTestCase, TestUtils
+
+
+class InterpolateTest(PandasOnSparkTestCase, TestUtils):
Review Comment:
Maybe you can create a file with broader scope. e.g., `test_generic_functions.py`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xinrong-databricks commented on a diff in pull request #36127: [SPARK-38844][PYTHON][SQL] Implement linear interpolate
Posted by GitBox <gi...@apache.org>.
xinrong-databricks commented on code in PR #36127:
URL: https://github.com/apache/spark/pull/36127#discussion_r847779346
##########
python/pyspark/pandas/frame.py:
##########
@@ -5500,6 +5500,25 @@ def op(psser: ps.Series) -> ps.Series:
else:
return psdf
+ def interpolate(
+ self, method: Optional[str] = None, limit: Optional[int] = None
+ ) -> Optional["DataFrame"]:
+ if (method is not None) and (method not in ["linear"]):
+ raise ValueError("Expecting 'linear'.")
Review Comment:
How about raising `NotImplementedError` for methods that pandas support?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org