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 2021/10/07 11:18:04 UTC
[GitHub] [spark] dchvn opened a new pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
dchvn opened a new pull request #34212:
URL: https://github.com/apache/spark/pull/34212
### What changes were proposed in this pull request?
Implement Series.combine
### Why are the changes needed?
Increase pandas API coverage in PySpark
### Does this PR introduce _any_ user-facing change?
User can use
```python
>>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
>>> s1
falcon 330.0
eagle 160.0
dtype: float64
>>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
>>> s2
falcon 345.0
eagle 200.0
duck 30.0
dtype: float64
>>> s1.combine(s2, max)
duck NaN
eagle 200.0
falcon 345.0
dtype: float64
```
### How was this patch tested?
unit tests and docstest
--
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] dchvn commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937897925
CC @ueshin @HyukjinKwon @itholic @xinrong-databricks , FYI
--
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 change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
itholic commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r769178179
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ assert callable(func), "argument func must be a callable function."
+
+ if np.isscalar(other):
Review comment:
ditto
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-995405500
**[Test build #146257 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/146257/testReport)** for PR 34212 at commit [`b42afcf`](https://github.com/apache/spark/commit/b42afcf28c2dfd7828578734a14125da1298190a).
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-995434066
Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/50729/
--
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] AmplabJenkins commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-995427124
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/146257/
--
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] dchvn commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r785753384
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
Review comment:
Thank you. I'm having trouble with the `infer_return_type` at https://github.com/apache/spark/pull/35200#issuecomment-1014263771
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r782668819
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
Review comment:
quick question: What if same_anchor with `fill_value`? see also my comments below test about `fill_value`.
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r793351769
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,181 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ This API does not support self combine for now.
+
+ >>> psser1 = ps.Series([1, 2, 3, 4])
+ >>> psser1.combine(psser1, max) # doctest: +SKIP
+ ...
+ ValueError: Unsupported self combine
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not callable(func):
+ raise TypeError("'%s' object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other): # type: ignore[attr-defined]
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ else:
+ assert isinstance(other, Series)
+ if same_anchor(self, other):
+ if self._column_label == other._column_label:
+ raise ValueError("Unsupported self combine")
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except (ValueError, TypeError):
Review comment:
Sorry, typo in above comment, updated:
```
If the type hints is not specified for the `func` parameter of `series.combine`
```
--
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] dchvn commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r785745008
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
Review comment:
I put `fill_value` tests in python/pyspark/pandas/tests/test_ops_on_diff_frames.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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r785687385
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
Review comment:
Thanks for clarify for `same_anchor with fill_value`, so we'd better also change the below test to cover this specific case, see it work or not?
Ah, I mean `when an index is missing from one Series or the other`, rather than `same_anchor with fill_value`, which is mentioned in below 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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937718684
**[Test build #143969 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/143969/testReport)** for PR 34212 at commit [`2cd0af1`](https://github.com/apache/spark/commit/2cd0af1b5c2f0e54669398eab53e4254fe3239ce).
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983285372
**[Test build #145793 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/145793/testReport)** for PR 34212 at commit [`d596472`](https://github.com/apache/spark/commit/d596472d95c72e689d6d0cec777c5adc0e46ee4c).
* This patch **fails PySpark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
--
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] SparkQA removed a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983280331
**[Test build #145793 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/145793/testReport)** for PR 34212 at commit [`d596472`](https://github.com/apache/spark/commit/d596472d95c72e689d6d0cec777c5adc0e46ee4c).
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983373300
**[Test build #145803 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/145803/testReport)** for PR 34212 at commit [`d596472`](https://github.com/apache/spark/commit/d596472d95c72e689d6d0cec777c5adc0e46ee4c).
--
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] AmplabJenkins removed a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983362682
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/50266/
--
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] AmplabJenkins commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983362682
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/50266/
--
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] AmplabJenkins commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983301304
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/145793/
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-995416109
**[Test build #146257 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/146257/testReport)** for PR 34212 at commit [`b42afcf`](https://github.com/apache/spark/commit/b42afcf28c2dfd7828578734a14125da1298190a).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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] AmplabJenkins commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983498759
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/50277/
--
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] dchvn commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r771057964
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ assert callable(func), "argument func must be a callable function."
+
+ if np.isscalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except TypeError:
+ limit = ps.get_option("compute.shortcut_limit")
+ pdf = combined.head(limit + 1)._to_internal_pandas()
+ combined_pser = pdf.iloc[:, 0].combine(pdf.iloc[:, 1], func, fill_value=fill_value)
+
+ return_dtype = combined_pser.dtype
+ return_spark_type = as_spark_type(return_dtype)
+
+ @pandas_udf(returnType=return_spark_type) # type: ignore
+ def wrapped_func(x: pd.Series, y: pd.Series) -> pd.Series:
+ return x.combine(y, func)
+
+ scol = wrapped_func(*combined._internal.data_spark_columns).alias(
+ self._internal.spark_column_name_for(self.spark.column)
+ )
+
+ data_field = self._internal.data_fields[0].copy(
+ dtype=return_dtype, spark_type=return_spark_type, nullable=True
Review comment:
hmm, the series `self` here has `nullable = False` and the `scol` output by `pandas_udf` has `nullable=True`. I have checked but not sure when a series which output by `pandas_udf` has `nullable` is `True` or `False`. Whether it is always True? @itholic @HyukjinKwon Do you know about 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] AmplabJenkins removed a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-995434078
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/50729/
--
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] dchvn commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r785674719
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
Review comment:
Yeah, I think the type hints is enough. Thanks!
--
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] dchvn commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r789395343
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
Review comment:
updated with ValueError exception
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937748535
**[Test build #143969 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/143969/testReport)** for PR 34212 at commit [`2cd0af1`](https://github.com/apache/spark/commit/2cd0af1b5c2f0e54669398eab53e4254fe3239ce).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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] dchvn commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983371360
Jenkins, retest this please
--
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] SparkQA removed a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937718684
**[Test build #143969 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/143969/testReport)** for PR 34212 at commit [`2cd0af1`](https://github.com/apache/spark/commit/2cd0af1b5c2f0e54669398eab53e4254fe3239ce).
--
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] SparkQA removed a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937718684
--
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] dchvn commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r770184266
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ assert callable(func), "argument func must be a callable function."
+
+ if np.isscalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except TypeError:
+ limit = ps.get_option("compute.shortcut_limit")
+ pdf = combined.head(limit + 1)._to_internal_pandas()
+ combined_pser = pdf.iloc[:, 0].combine(pdf.iloc[:, 1], func, fill_value=fill_value)
+
+ return_dtype = combined_pser.dtype
+ return_spark_type = as_spark_type(return_dtype)
+
+ @pandas_udf(returnType=return_spark_type) # type: ignore
+ def wrapped_func(x: pd.Series, y: pd.Series) -> pd.Series:
+ return x.combine(y, func)
+
+ scol = wrapped_func(*combined._internal.data_spark_columns).alias(
+ self._internal.spark_column_name_for(self.spark.column)
+ )
+
+ data_field = self._internal.data_fields[0].copy(
+ dtype=return_dtype, spark_type=return_spark_type, nullable=True
Review comment:
updated, thanks!
--
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 change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
itholic commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r769186662
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ assert callable(func), "argument func must be a callable function."
+
+ if np.isscalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
Review comment:
I think we also handle the case `same_anchor` with `fill_value is not None` ??
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ assert callable(func), "argument func must be a callable function."
+
+ if np.isscalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
Review comment:
I think we also handle the case `same_anchor` and `fill_value is not None` ??
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r785687385
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
Review comment:
Thanks for clarify, so we'd better also change the below test to cover this specific case, see it work or not?
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r785691941
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
Review comment:
And if it's not very easy to solve, we'd better to do some investigation, and raise a jira for reference.
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r782672043
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
Review comment:
No, I think it should be required.
```
>>> pdf["s1"].combine(pdf["s2"])
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
TypeError: combine() missing 1 required positional argument: 'func'
```
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937760278
**[Test build #143983 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/143983/testReport)** for PR 34212 at commit [`2cd0af1`](https://github.com/apache/spark/commit/2cd0af1b5c2f0e54669398eab53e4254fe3239ce).
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937878905
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/48461/
--
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] AmplabJenkins removed a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937759903
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/143969/
--
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 change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
itholic commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r769177110
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
Review comment:
Shall we use `is_scalar` from `pandas.api.types`, rather than `np.isscalar` ??
Seems like pandas' can cover more Python scalar types including NumPy array scalar.
pandas `is_scalar`: https://pandas.pydata.org/docs/reference/api/pandas.api.types.is_scalar.html
NumPy `isscalar`: https://numpy.org/doc/stable/reference/generated/numpy.isscalar.html
--
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] AmplabJenkins commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-995434078
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/50729/
--
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 change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
itholic commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r769178179
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ assert callable(func), "argument func must be a callable function."
+
+ if np.isscalar(other):
Review comment:
ditto
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r793348605
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,190 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def multiply(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ This API does not support self combine for now.
+
+ >>> psser1 = ps.Series([1, 2, 3, 4])
+ >>> psser1.combine(psser1, max) # doctest: +SKIP
+ ...
+ ValueError: Unsupported self combine
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> def max_with_return_type(x, y) -> float:
+ ... return max(x, y)
+ ...
+ >>> s1.combine(s2, max_with_return_type)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max_with_return_type, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not callable(func):
+ raise TypeError("'%s' object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other): # type: ignore[attr-defined]
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ else:
+ assert isinstance(other, Series)
+ if same_anchor(self, other):
+ if self._column_label == other._column_label:
+ raise ValueError("Unsupported self combine")
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except (ValueError, TypeError):
+ # Here we execute with the first 1000 to get the return type.
+ # If the records were less than 1000, it uses pandas API directly for a shortcut.
+ log_advice(
+ "If the type hints is not specified the `func` parameter for `series.combine`, "
Review comment:
Sorry, wrong type in my above comments.
`If the type hints is not specified for the `func` parameter of `series.combine`,`
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,190 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def multiply(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ This API does not support self combine for now.
+
+ >>> psser1 = ps.Series([1, 2, 3, 4])
+ >>> psser1.combine(psser1, max) # doctest: +SKIP
+ ...
+ ValueError: Unsupported self combine
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> def max_with_return_type(x, y) -> float:
+ ... return max(x, y)
+ ...
+ >>> s1.combine(s2, max_with_return_type)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max_with_return_type, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not callable(func):
+ raise TypeError("'%s' object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other): # type: ignore[attr-defined]
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ else:
+ assert isinstance(other, Series)
+ if same_anchor(self, other):
+ if self._column_label == other._column_label:
+ raise ValueError("Unsupported self combine")
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except (ValueError, TypeError):
+ # Here we execute with the first 1000 to get the return type.
+ # If the records were less than 1000, it uses pandas API directly for a shortcut.
+ log_advice(
+ "If the type hints is not specified the `func` parameter for `series.combine`, "
Review comment:
Sorry, typo in my above comments.
`If the type hints is not specified for the `func` parameter of `series.combine`,`
--
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] dchvn commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-1022957434
CC @HyukjinKwon Could please take a look?
--
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] AmplabJenkins commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983402466
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/145803/
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983446971
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/50277/
--
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] dchvn commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r785786499
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
Review comment:
> And if it's not very easy to solve, we'd better to do some investigation, and raise a jira for reference.
I filed a ticket https://issues.apache.org/jira/browse/SPARK-37930 for it. Thanks!
--
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] AmplabJenkins removed a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937933307
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/48461/
--
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] dchvn commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r794995636
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,190 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def multiply(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ This API does not support self combine for now.
+
+ >>> psser1 = ps.Series([1, 2, 3, 4])
+ >>> psser1.combine(psser1, max) # doctest: +SKIP
+ ...
+ ValueError: Unsupported self combine
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> def max_with_return_type(x, y) -> float:
+ ... return max(x, y)
+ ...
+ >>> s1.combine(s2, max_with_return_type)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max_with_return_type, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not callable(func):
+ raise TypeError("'%s' object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other): # type: ignore[attr-defined]
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ else:
+ assert isinstance(other, Series)
+ if same_anchor(self, other):
+ if self._column_label == other._column_label:
+ raise ValueError("Unsupported self combine")
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except (ValueError, TypeError):
+ # Here we execute with the first 1000 to get the return type.
+ # If the records were less than 1000, it uses pandas API directly for a shortcut.
+ log_advice(
+ "If the type hints is not specified the `func` parameter for `series.combine`, "
+ "it is expensive to infer the data type internally."
Review comment:
Oh, thank you!
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r793219057
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,181 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ This API does not support self combine for now.
+
+ >>> psser1 = ps.Series([1, 2, 3, 4])
+ >>> psser1.combine(psser1, max) # doctest: +SKIP
+ ...
+ ValueError: Unsupported self combine
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
Review comment:
but this is not a best practice for `the good note`, let's give the right example like:
```
>>> def max_with_return_type(x, y) -> float:
... return max(x, y)
>>> s1.combine(s2, max)
```
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,181 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
Review comment:
nits:
```suggestion
>>> def multiply(x, y) -> np.int32:
```
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,181 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ This API does not support self combine for now.
+
+ >>> psser1 = ps.Series([1, 2, 3, 4])
+ >>> psser1.combine(psser1, max) # doctest: +SKIP
+ ...
+ ValueError: Unsupported self combine
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
Review comment:
ditto
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,181 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ This API does not support self combine for now.
+
+ >>> psser1 = ps.Series([1, 2, 3, 4])
+ >>> psser1.combine(psser1, max) # doctest: +SKIP
+ ...
+ ValueError: Unsupported self combine
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not callable(func):
+ raise TypeError("'%s' object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other): # type: ignore[attr-defined]
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ else:
+ assert isinstance(other, Series)
+ if same_anchor(self, other):
+ if self._column_label == other._column_label:
+ raise ValueError("Unsupported self combine")
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except (ValueError, TypeError):
Review comment:
Let's add note and log_advice in here, like:
```suggestion
except (ValueError, TypeError):
# Here we execute with the first 1000 to get the return type.
# If the records were less than 1000, it uses pandas API directly for a shortcut.
log_advice(
"If the type hints is not specified the `func` parameter for `series.combine`, "
"it is expensive to infer the data type internally."
)
```
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,181 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
Review comment:
Yep, this is a good note
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,181 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ This API does not support self combine for now.
+
+ >>> psser1 = ps.Series([1, 2, 3, 4])
+ >>> psser1.combine(psser1, max) # doctest: +SKIP
+ ...
+ ValueError: Unsupported self combine
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not callable(func):
+ raise TypeError("'%s' object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other): # type: ignore[attr-defined]
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ else:
+ assert isinstance(other, Series)
+ if same_anchor(self, other):
+ if self._column_label == other._column_label:
+ raise ValueError("Unsupported self combine")
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
Review comment:
nits: no regresson test on this, but if you update the doctest as my suggestion, It will have.
--
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] dchvn commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-1021769576
@itholic @Yikun Could you please take another look when you find sometime?
--
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 change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
itholic commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r769178024
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ assert callable(func), "argument func must be a callable function."
Review comment:
pandas raises `TypeError` for wrong value for `func`.
We should match the behavior with pandas.
```python
>>> pser1.combine(pser2, 10)
Traceback (most recent call last):
...
TypeError: 'int' object is not callable
```
--
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 change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
itholic commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r769188440
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
Review comment:
I think it's not "required", but "strongly recommended" ?
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
Review comment:
Shall we use is_scalar from pandas.api.types, rather than np.isscalar ??
Seems like pandas' can cover more Python scalar types including NumPy array scalar.
pandas is_scalar: https://pandas.pydata.org/docs/reference/api/pandas.api.types.is_scalar.html
NumPy isscalar: https://numpy.org/doc/stable/reference/generated/numpy.isscalar.html
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ assert callable(func), "argument func must be a callable function."
Review comment:
pandas raises `TypeError` for wrong value for `func`.
We should match the behavior with pandas.
```python
>>> pser1.combine(pser2, 10)
Traceback (most recent call last):
...
TypeError: 'int' object is not callable
```
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ assert callable(func), "argument func must be a callable function."
+
+ if np.isscalar(other):
Review comment:
ditto
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ assert callable(func), "argument func must be a callable function."
+
+ if np.isscalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except TypeError:
+ limit = ps.get_option("compute.shortcut_limit")
+ pdf = combined.head(limit + 1)._to_internal_pandas()
+ combined_pser = pdf.iloc[:, 0].combine(pdf.iloc[:, 1], func, fill_value=fill_value)
+
+ return_dtype = combined_pser.dtype
+ return_spark_type = as_spark_type(return_dtype)
+
+ @pandas_udf(returnType=return_spark_type) # type: ignore
+ def wrapped_func(x: pd.Series, y: pd.Series) -> pd.Series:
+ return x.combine(y, func)
+
+ scol = wrapped_func(*combined._internal.data_spark_columns).alias(
+ self._internal.spark_column_name_for(self.spark.column)
+ )
+
+ data_field = self._internal.data_fields[0].copy(
+ dtype=return_dtype, spark_type=return_spark_type, nullable=True
+ )
+
+ internal = combined._internal.copy(
+ data_spark_columns=[scol],
+ data_fields=[data_field],
+ column_labels=self._internal.column_labels,
+ column_label_names=self._internal.column_label_names,
+ )
+
+ result = first_series(DataFrame(internal))
+ if isinstance(other, Series) and (self.name != other.name):
+ result.name = None
+
+ return result
+
+ def _combine_frame_with_fill_value(self, other: "Series", fill_value: Any) -> DataFrame:
+ this = self.to_frame()
+ this_col = this._internal.column_labels[0]
+
+ that = other.to_frame()
+ that_col = that._internal.column_labels[0]
+
+ tmp_col = "__tmp_check_missing_index__"
+ verify_temp_column_name(this, tmp_col)
+ verify_temp_column_name(that, tmp_col)
+ this[tmp_col], that[tmp_col] = True, True
+
+ combined = combine_frames(this, that)
+
+ combined.loc[combined[("this", tmp_col)].isnull(), ("this",) + this_col] = fill_value
+ combined.loc[combined[("that", tmp_col)].isnull(), ("that",) + that_col] = fill_value
+
+ return combined[[("this",) + this_col, ("that",) + that_col]]
Review comment:
Can we make `("this",) + this_col` and `("that",) + that_col` as variables and reuse it here and there ??
e.g.
```python
this_col_without_tmp = ("this",) + this_col
that_col_without_tmp = ("that",) + that_col
```
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ assert callable(func), "argument func must be a callable function."
+
+ if np.isscalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except TypeError:
+ limit = ps.get_option("compute.shortcut_limit")
+ pdf = combined.head(limit + 1)._to_internal_pandas()
+ combined_pser = pdf.iloc[:, 0].combine(pdf.iloc[:, 1], func, fill_value=fill_value)
+
+ return_dtype = combined_pser.dtype
+ return_spark_type = as_spark_type(return_dtype)
+
+ @pandas_udf(returnType=return_spark_type) # type: ignore
+ def wrapped_func(x: pd.Series, y: pd.Series) -> pd.Series:
+ return x.combine(y, func)
+
+ scol = wrapped_func(*combined._internal.data_spark_columns).alias(
+ self._internal.spark_column_name_for(self.spark.column)
+ )
+
+ data_field = self._internal.data_fields[0].copy(
+ dtype=return_dtype, spark_type=return_spark_type, nullable=True
Review comment:
qq: why we always set the nullable as `True` ??
Maybe we should just copy the existing one ??
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-995411646
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/50729/
--
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 change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
itholic commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r769178024
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ assert callable(func), "argument func must be a callable function."
Review comment:
pandas raises `TypeError` for wrong value for `func`.
We should match the behavior with pandas.
```python
>>> pser1.combine(pser2, 10)
Traceback (most recent call last):
...
TypeError: 'int' object is not callable
```
--
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 #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-988500748
cc @itholic @xinrong-databricks @Yikun would you mind reviewing this when you find some time please 🙏
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983320773
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/50266/
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983395095
**[Test build #145803 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/145803/testReport)** for PR 34212 at commit [`d596472`](https://github.com/apache/spark/commit/d596472d95c72e689d6d0cec777c5adc0e46ee4c).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r782649085
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
Review comment:
`other` can be `Series` or scalar in here, so it should be `Union[Scalar, "Series"]`?
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r782652495
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
Review comment:
nit: It reasonable, but I guess It raised the different exception with pandas. If we still need this, at least, add a note on doc to say it's the different behavior with pandas but we do the right thing maybe? or the type hints is enough in here.
and "Unsupported"
--
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] AmplabJenkins commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937759903
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/143969/
--
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] SparkQA removed a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937760278
**[Test build #143983 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/143983/testReport)** for PR 34212 at commit [`2cd0af1`](https://github.com/apache/spark/commit/2cd0af1b5c2f0e54669398eab53e4254fe3239ce).
--
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] AmplabJenkins commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937759903
--
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] AmplabJenkins removed a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937759903
--
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] dchvn commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r789395428
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
Review comment:
updated
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r793220795
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,181 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ This API does not support self combine for now.
+
+ >>> psser1 = ps.Series([1, 2, 3, 4])
+ >>> psser1.combine(psser1, max) # doctest: +SKIP
+ ...
+ ValueError: Unsupported self combine
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not callable(func):
+ raise TypeError("'%s' object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other): # type: ignore[attr-defined]
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ else:
+ assert isinstance(other, Series)
+ if same_anchor(self, other):
+ if self._column_label == other._column_label:
+ raise ValueError("Unsupported self combine")
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except (ValueError, TypeError):
Review comment:
Let's add note and log_advice in here, like:
```suggestion
except (ValueError, TypeError):
# Here we execute with the first 1000 to get the return type.
# If the records were less than 1000, it uses pandas API directly for a shortcut.
log_advice(
"If the type hints is not specified for the `func` parameter of `series.combine`, "
"it is expensive to infer the data type internally."
)
```
--
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] dchvn edited a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn edited a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-1022957434
CC @HyukjinKwon Could you please take a look?
--
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] AmplabJenkins removed a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-995427124
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/146257/
--
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] AmplabJenkins removed a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983498759
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/50277/
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983280331
**[Test build #145793 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/145793/testReport)** for PR 34212 at commit [`d596472`](https://github.com/apache/spark/commit/d596472d95c72e689d6d0cec777c5adc0e46ee4c).
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r785687385
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
Review comment:
Thanks for clarify for `same_anchor with fill_value`, so we'd better also change the below test to cover this specific case, see it work or not?
Ah, I mean `when an index is missing from one Series or the other`, rather than `same_anchor with fill_value`
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937933278
Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/48461/
--
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] dchvn commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-939604224
CC @ueshin, Could you take another look ? Thank you so much!
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937718684
--
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] dchvn commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-1048641943
Ping @itholic , @Yikun , @HyukjinKwon , Hope you have a look again! Thanks in advance!
--
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 change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
itholic commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r769177110
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
Review comment:
Shall we use `is_scalar` from `pandas.api.types`, rather than `np.isscalar` ??
Seems like pandas' can cover more Python scalar types including NumPy array scalar.
pandas `is_scalar`: https://pandas.pydata.org/docs/reference/api/pandas.api.types.is_scalar.html
NumPy `isscalar`: https://numpy.org/doc/stable/reference/generated/numpy.isscalar.html
--
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] dchvn commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r770184104
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
Review comment:
make sense! thank you 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] itholic commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
itholic commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r769186662
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ assert callable(func), "argument func must be a callable function."
+
+ if np.isscalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
Review comment:
How about `same_anchor` with `fill_value is not None` ??
--
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] SparkQA removed a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-995405500
**[Test build #146257 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/146257/testReport)** for PR 34212 at commit [`b42afcf`](https://github.com/apache/spark/commit/b42afcf28c2dfd7828578734a14125da1298190a).
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983341397
Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/50266/
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937789446
**[Test build #143983 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/143983/testReport)** for PR 34212 at commit [`2cd0af1`](https://github.com/apache/spark/commit/2cd0af1b5c2f0e54669398eab53e4254fe3239ce).
* This patch **fails PySpark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r782649085
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
Review comment:
`other` can be `Series` or scalar in here, so I guess it should be `Union[Scalar, "Series"]`?
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r793348605
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,190 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def multiply(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ This API does not support self combine for now.
+
+ >>> psser1 = ps.Series([1, 2, 3, 4])
+ >>> psser1.combine(psser1, max) # doctest: +SKIP
+ ...
+ ValueError: Unsupported self combine
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> def max_with_return_type(x, y) -> float:
+ ... return max(x, y)
+ ...
+ >>> s1.combine(s2, max_with_return_type)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max_with_return_type, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not callable(func):
+ raise TypeError("'%s' object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other): # type: ignore[attr-defined]
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ else:
+ assert isinstance(other, Series)
+ if same_anchor(self, other):
+ if self._column_label == other._column_label:
+ raise ValueError("Unsupported self combine")
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except (ValueError, TypeError):
+ # Here we execute with the first 1000 to get the return type.
+ # If the records were less than 1000, it uses pandas API directly for a shortcut.
+ log_advice(
+ "If the type hints is not specified the `func` parameter for `series.combine`, "
Review comment:
Sorry, typo in my above comments.
```
If the type hints is not specified for the `func` parameter of `series.combine`,
```
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r793351769
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,181 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ This API does not support self combine for now.
+
+ >>> psser1 = ps.Series([1, 2, 3, 4])
+ >>> psser1.combine(psser1, max) # doctest: +SKIP
+ ...
+ ValueError: Unsupported self combine
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not callable(func):
+ raise TypeError("'%s' object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other): # type: ignore[attr-defined]
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ else:
+ assert isinstance(other, Series)
+ if same_anchor(self, other):
+ if self._column_label == other._column_label:
+ raise ValueError("Unsupported self combine")
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except (ValueError, TypeError):
Review comment:
If the type hints is not specified for the `func` parameter of `series.combine`
--
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] AmplabJenkins commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937933307
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/48461/
--
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] dchvn commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r771057964
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ assert callable(func), "argument func must be a callable function."
+
+ if np.isscalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except TypeError:
+ limit = ps.get_option("compute.shortcut_limit")
+ pdf = combined.head(limit + 1)._to_internal_pandas()
+ combined_pser = pdf.iloc[:, 0].combine(pdf.iloc[:, 1], func, fill_value=fill_value)
+
+ return_dtype = combined_pser.dtype
+ return_spark_type = as_spark_type(return_dtype)
+
+ @pandas_udf(returnType=return_spark_type) # type: ignore
+ def wrapped_func(x: pd.Series, y: pd.Series) -> pd.Series:
+ return x.combine(y, func)
+
+ scol = wrapped_func(*combined._internal.data_spark_columns).alias(
+ self._internal.spark_column_name_for(self.spark.column)
+ )
+
+ data_field = self._internal.data_fields[0].copy(
+ dtype=return_dtype, spark_type=return_spark_type, nullable=True
Review comment:
hmm, the series `self` here has `nullable = False` and the `scol` output by `pandas_udf` has `nullable=True`. I have checked but not sure when a series which output by `pandas_udf` has `nullable` is `True` or `False`. Whether it is always True?
--
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 change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
itholic commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r769173394
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
Review comment:
I think it's not "required", but "strongly recommended" ?
--
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] AmplabJenkins removed a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983301304
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/145793/
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r793351769
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,181 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ This API does not support self combine for now.
+
+ >>> psser1 = ps.Series([1, 2, 3, 4])
+ >>> psser1.combine(psser1, max) # doctest: +SKIP
+ ...
+ ValueError: Unsupported self combine
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not callable(func):
+ raise TypeError("'%s' object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other): # type: ignore[attr-defined]
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ else:
+ assert isinstance(other, Series)
+ if same_anchor(self, other):
+ if self._column_label == other._column_label:
+ raise ValueError("Unsupported self combine")
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except (ValueError, TypeError):
Review comment:
Sorry, typo in above comment:
```
If the type hints is not specified for the `func` parameter of `series.combine`
```
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r793351769
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,181 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ This API does not support self combine for now.
+
+ >>> psser1 = ps.Series([1, 2, 3, 4])
+ >>> psser1.combine(psser1, max) # doctest: +SKIP
+ ...
+ ValueError: Unsupported self combine
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not callable(func):
+ raise TypeError("'%s' object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other): # type: ignore[attr-defined]
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ else:
+ assert isinstance(other, Series)
+ if same_anchor(self, other):
+ if self._column_label == other._column_label:
+ raise ValueError("Unsupported self combine")
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except (ValueError, TypeError):
Review comment:
@dchvn Sorry, typo in above comment, updated:
```
If the type hints is not specified for the `func` parameter of `series.combine`
```
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r793225136
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,181 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
Review comment:
or maybe just give a `max` exmaple, it would be fluent when user see below doctest.
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r782649085
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
Review comment:
other can be `Series` or scalar in here, so I guess it should be `Union[Scalar, "Series"]`?
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
Review comment:
It reasonable, but I guess It raised the different exception with pandas.
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
Review comment:
Looks like there are some indexing problem on self combined.
Try this case:
```
self.assert_eq(pdf["s1"].combine(pdf["s1"], max), psdf["s1"].combine(psdf["s1"], max))
```
##########
File path: python/pyspark/pandas/tests/test_series.py
##########
@@ -3020,6 +3020,67 @@ def test_eq(self):
with self.assertRaisesRegex(ValueError, "Lengths must be equal"):
psser == other
+ def test_combine(self):
+ pdf = pd.DataFrame(
+ {"s1": [330.0, 160.0, np.nan], "s2": [345.0, 0.0, 30.0], "s3": [345.0, 0.0, 30.0]}
+ )
+ psdf = ps.from_pandas(pdf)
+
+ self.assert_eq(
+ pdf["s1"].combine(pdf["s2"], max),
+ psdf["s1"].combine(psdf["s2"], max),
+ )
+ self.assert_eq(
+ pdf["s1"].combine(pdf["s2"], max, fill_value=100),
Review comment:
Question: Does fill_value work in pandas actually? Looks like we need to find a real case to fill the value I guess.
```
>>> pdf["s1"].combine(pdf["s2"], max, fill_value=100)
0 345.0
1 160.0
2 NaN
dtype: float64
```
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
Review comment:
```suggestion
raise TypeError("'%s' object is not callable" % type(func).__name__)
```
Follow pandas exception format maybe.
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
Review comment:
quick question: What if same_anchor with `fill_value`?
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
Review comment:
nit: It reasonable, but I guess It raised the different exception with pandas. If we still need this, at least, add a note on it to say it's the different behavior with pandas but we do the right thing maybe?
and "Unsupported"
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r782669292
##########
File path: python/pyspark/pandas/tests/test_series.py
##########
@@ -3020,6 +3020,67 @@ def test_eq(self):
with self.assertRaisesRegex(ValueError, "Lengths must be equal"):
psser == other
+ def test_combine(self):
+ pdf = pd.DataFrame(
+ {"s1": [330.0, 160.0, np.nan], "s2": [345.0, 0.0, 30.0], "s3": [345.0, 0.0, 30.0]}
+ )
+ psdf = ps.from_pandas(pdf)
+
+ self.assert_eq(
+ pdf["s1"].combine(pdf["s2"], max),
+ psdf["s1"].combine(psdf["s2"], max),
+ )
+ self.assert_eq(
+ pdf["s1"].combine(pdf["s2"], max, fill_value=100),
Review comment:
Question: Does `fill_value` work in pandas actually? Looks like we need to find a real case to fill the value I guess.
```
>>> pdf["s1"].combine(pdf["s2"], max, fill_value=100)
0 345.0
1 160.0
2 NaN
dtype: float64
```
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r785687385
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
Review comment:
Thanks for clarify for `same_anchor with fill_value`, so we'd better also change the below test to cover this specific case, see it work or not?
Ah, I mean `when an index is missing from one Series or the other`, rather than `same_anchor with fill_value`, which was mentioned in below 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] dchvn commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r789395729
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
Review comment:
updated, thanks
--
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] dchvn commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r789401681
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not np.isscalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ assert callable(func), "argument func must be a callable function."
+
+ if np.isscalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except TypeError:
+ limit = ps.get_option("compute.shortcut_limit")
+ pdf = combined.head(limit + 1)._to_internal_pandas()
+ combined_pser = pdf.iloc[:, 0].combine(pdf.iloc[:, 1], func, fill_value=fill_value)
+
+ return_dtype = combined_pser.dtype
+ return_spark_type = as_spark_type(return_dtype)
+
+ @pandas_udf(returnType=return_spark_type) # type: ignore
+ def wrapped_func(x: pd.Series, y: pd.Series) -> pd.Series:
+ return x.combine(y, func)
+
+ scol = wrapped_func(*combined._internal.data_spark_columns).alias(
+ self._internal.spark_column_name_for(self.spark.column)
+ )
+
+ data_field = self._internal.data_fields[0].copy(
+ dtype=return_dtype, spark_type=return_spark_type, nullable=True
Review comment:
I reverted to keep `nullable=True`
--
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 change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
itholic commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r769188440
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
Review comment:
I think it should not be "required", but "strongly recommended" ?
--
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 change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
itholic commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r769173394
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
Review comment:
I think it's not "required", but "strongly recommended" ?
--
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 #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-1010513661
cc @itholic can you review this please
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r782652495
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
Review comment:
nit: It reasonable, but I guess It raised the different exception with pandas. If we still need this, at least, add a note on it to say it's the different behavior with pandas but we do the right thing maybe? or the type hints is enough in here.
and "Unsupported"
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r782672473
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,170 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is required.
Review comment:
oh, sorry, it's said for type hint, just ignore my 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] AmplabJenkins removed a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983402466
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/145803/
--
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] SparkQA commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983482102
Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/50277/
--
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] SparkQA removed a comment on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-983373300
**[Test build #145803 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/145803/testReport)** for PR 34212 at commit [`d596472`](https://github.com/apache/spark/commit/d596472d95c72e689d6d0cec777c5adc0e46ee4c).
--
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] dchvn commented on pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on pull request #34212:
URL: https://github.com/apache/spark/pull/34212#issuecomment-937897925
CC @ueshin @HyukjinKwon @itholic @xinrong-databricks , FYI
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r794989958
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4483,6 +4487,190 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: Union[Scalar, "Series"],
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: This API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def multiply(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ This API does not support self combine for now.
+
+ >>> psser1 = ps.Series([1, 2, 3, 4])
+ >>> psser1.combine(psser1, max) # doctest: +SKIP
+ ...
+ ValueError: Unsupported self combine
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> def max_with_return_type(x, y) -> float:
+ ... return max(x, y)
+ ...
+ >>> s1.combine(s2, max_with_return_type)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max_with_return_type, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not callable(func):
+ raise TypeError("'%s' object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other): # type: ignore[attr-defined]
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ else:
+ assert isinstance(other, Series)
+ if same_anchor(self, other):
+ if self._column_label == other._column_label:
+ raise ValueError("Unsupported self combine")
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
+ combined = combine_frames(self.to_frame(), other.to_frame())
+ else:
+ combined = self._combine_frame_with_fill_value(other, fill_value=fill_value)
+
+ try:
+ sig_return = infer_return_type(func)
+ if isinstance(sig_return, (UnknownType, DataFrameType)):
+ raise TypeError()
+ return_spark_type = sig_return.spark_type
+ return_dtype = sig_return.dtype
+ except (ValueError, TypeError):
+ # Here we execute with the first 1000 to get the return type.
+ # If the records were less than 1000, it uses pandas API directly for a shortcut.
+ log_advice(
+ "If the type hints is not specified the `func` parameter for `series.combine`, "
+ "it is expensive to infer the data type internally."
Review comment:
see https://github.com/apache/spark/pull/34212#discussion_r793351769 , : )
--
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] dchvn commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
dchvn commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r785677677
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
Review comment:
`fill_value` is only used when an index is missing from one Series or the other. With two series from same dataframe, there are no missing indices.
https://pandas.pydata.org/docs/reference/api/pandas.Series.combine.html
--
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] Yikun commented on a change in pull request #34212: [SPARK-36402][PYTHON] Implement Series.combine
Posted by GitBox <gi...@apache.org>.
Yikun commented on a change in pull request #34212:
URL: https://github.com/apache/spark/pull/34212#discussion_r785687385
##########
File path: python/pyspark/pandas/series.py
##########
@@ -4485,6 +4489,173 @@ def replace(
return self._with_new_scol(current) # TODO: dtype?
+ def combine(
+ self,
+ other: "Series",
+ func: Callable,
+ fill_value: Optional[Any] = None,
+ ) -> "Series":
+ """
+ Combine the Series with a Series or scalar according to `func`.
+
+ Combine the Series and `other` using `func` to perform elementwise
+ selection for combined Series.
+ `fill_value` is assumed when value is missing at some index
+ from one of the two objects being combined.
+
+ .. versionadded:: 3.3.0
+
+ .. note:: this API executes the function once to infer the type which is
+ potentially expensive, for instance, when the dataset is created after
+ aggregations or sorting.
+
+ To avoid this, specify return type in ``func``, for instance, as below:
+
+ >>> def foo(x, y) -> np.int32:
+ ... return x * y
+
+ pandas-on-Spark uses return type hint and does not try to infer the type.
+
+ Parameters
+ ----------
+ other : Series or scalar
+ The value(s) to be combined with the `Series`.
+ func : function
+ Function that takes two scalars as inputs and returns an element.
+ Note that type hint for return type is strongly recommended.
+ fill_value : scalar, optional
+ The value to assume when an index is missing from
+ one Series or the other. The default specifies to use the
+ appropriate NaN value for the underlying dtype of the Series.
+
+ Returns
+ -------
+ Series
+ The result of combining the Series with the other object.
+
+ See Also
+ --------
+ Series.combine_first : Combine Series values, choosing the calling
+ Series' values first.
+
+ Examples
+ --------
+ Consider 2 Datasets ``s1`` and ``s2`` containing
+ highest clocked speeds of different birds.
+
+ >>> from pyspark.pandas.config import set_option, reset_option
+ >>> set_option("compute.ops_on_diff_frames", True)
+ >>> s1 = ps.Series({'falcon': 330.0, 'eagle': 160.0})
+ >>> s1
+ falcon 330.0
+ eagle 160.0
+ dtype: float64
+ >>> s2 = ps.Series({'falcon': 345.0, 'eagle': 200.0, 'duck': 30.0})
+ >>> s2
+ falcon 345.0
+ eagle 200.0
+ duck 30.0
+ dtype: float64
+
+ Now, to combine the two datasets and view the highest speeds
+ of the birds across the two datasets
+
+ >>> s1.combine(s2, max)
+ duck NaN
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+
+ In the previous example, the resulting value for duck is missing,
+ because the maximum of a NaN and a float is a NaN.
+ So, in the example, we set ``fill_value=0``,
+ so the maximum value returned will be the value from some dataset.
+
+ >>> s1.combine(s2, max, fill_value=0)
+ duck 30.0
+ eagle 200.0
+ falcon 345.0
+ dtype: float64
+ >>> reset_option("compute.ops_on_diff_frames")
+ """
+ if not isinstance(other, Series) and not pd.api.types.is_scalar(other):
+ raise TypeError("unsupported type: %s" % type(other))
+
+ if not callable(func):
+ raise TypeError("%s object is not callable" % type(func).__name__)
+
+ if pd.api.types.is_scalar(other):
+ tmp_other_col = verify_temp_column_name(self._internal.spark_frame, "__tmp_other_col__")
+ combined = self.to_frame()
+ combined[tmp_other_col] = other
+ combined = DataFrame(combined._internal.resolved_copy)
+ elif same_anchor(self, other):
+ combined = self._psdf[self._column_label, other._column_label]
+ elif fill_value is None:
Review comment:
Thanks for clarify, so we'd better also change the below test to cover this specific case, see it work or not?
Ah, I mean `when an index is missing from one Series or the other`, rather than `same_anchor with fill_value`
--
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