You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "itholic (via GitHub)" <gi...@apache.org> on 2023/06/20 06:32:08 UTC
[GitHub] [spark] itholic opened a new pull request, #41670: [SPARK-43631][CONNECT][PS] Enable Series.interpolate with Spark Connect
itholic opened a new pull request, #41670:
URL: https://github.com/apache/spark/pull/41670
### What changes were proposed in this pull request?
This PR proposes to add `LastNonNull` and `NullIndex` to SparkConnectPlanner to enable `Series.interpolate`.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
To increase pandas API coverage
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
Yes, `Series.interpolate` will be available from this fix.
### How was this patch tested?
Reusing the existing UT.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] itholic commented on a diff in pull request #41670: [SPARK-43631][CONNECT][PS] Enable Series.interpolate with Spark Connect
Posted by "itholic (via GitHub)" <gi...@apache.org>.
itholic commented on code in PR #41670:
URL: https://github.com/apache/spark/pull/41670#discussion_r1238346198
##########
python/pyspark/pandas/series.py:
##########
@@ -2257,27 +2257,44 @@ def _interpolate(
return self._psdf.copy()._psser_for(self._column_label)
scol = self.spark.column
- sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
- last_non_null = PySparkColumn(sql_utils.lastNonNull(scol._jc))
- null_index = PySparkColumn(sql_utils.nullIndex(scol._jc))
+ if is_remote():
+ from pyspark.sql.connect.functions import _invoke_function_over_columns
Review Comment:
Thanks, @zhengruifeng ! Just adjusted the 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] zhengruifeng closed pull request #41670: [SPARK-43631][CONNECT][PS] Enable Series.interpolate with Spark Connect
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng closed pull request #41670: [SPARK-43631][CONNECT][PS] Enable Series.interpolate with Spark Connect
URL: https://github.com/apache/spark/pull/41670
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #41670: [WIP][SPARK-43631][CONNECT][PS] Enable Series.interpolate with Spark Connect
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #41670:
URL: https://github.com/apache/spark/pull/41670#discussion_r1234872865
##########
python/pyspark/pandas/series.py:
##########
@@ -2257,27 +2257,44 @@ def _interpolate(
return self._psdf.copy()._psser_for(self._column_label)
scol = self.spark.column
- sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
- last_non_null = PySparkColumn(sql_utils.lastNonNull(scol._jc))
- null_index = PySparkColumn(sql_utils.nullIndex(scol._jc))
+ if is_remote():
+ from pyspark.sql.connect.functions import _invoke_function_over_columns
+ last_non_null = _invoke_function_over_columns(
+ "last_non_null",
+ scol, # type: ignore[arg-type]
+ )
+ null_index = _invoke_function_over_columns(
+ "null_index",
+ scol, # type: ignore[arg-type]
+ )
+ else:
+ sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
+ last_non_null = PySparkColumn(
+ sql_utils.lastNonNull(scol._jc) # type: ignore[assignment]
+ )
+ null_index = PySparkColumn(sql_utils.nullIndex(scol._jc)) # type: ignore[assignment]
Review Comment:
let me take a look, no worries
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on pull request #41670: [SPARK-43631][CONNECT][PS] Enable Series.interpolate with Spark Connect
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on PR #41670:
URL: https://github.com/apache/spark/pull/41670#issuecomment-1609682165
@itholic it seems that need to resolve the conflicts now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] itholic commented on a diff in pull request #41670: [WIP][SPARK-43631][CONNECT][PS] Enable Series.interpolate with Spark Connect
Posted by "itholic (via GitHub)" <gi...@apache.org>.
itholic commented on code in PR #41670:
URL: https://github.com/apache/spark/pull/41670#discussion_r1236271460
##########
python/pyspark/pandas/series.py:
##########
@@ -2257,27 +2257,44 @@ def _interpolate(
return self._psdf.copy()._psser_for(self._column_label)
scol = self.spark.column
- sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
- last_non_null = PySparkColumn(sql_utils.lastNonNull(scol._jc))
- null_index = PySparkColumn(sql_utils.nullIndex(scol._jc))
+ if is_remote():
+ from pyspark.sql.connect.functions import _invoke_function_over_columns
+ last_non_null = _invoke_function_over_columns(
+ "last_non_null",
+ scol, # type: ignore[arg-type]
+ )
+ null_index = _invoke_function_over_columns(
+ "null_index",
+ scol, # type: ignore[arg-type]
+ )
+ else:
+ sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
+ last_non_null = PySparkColumn(
+ sql_utils.lastNonNull(scol._jc) # type: ignore[assignment]
+ )
+ null_index = PySparkColumn(sql_utils.nullIndex(scol._jc)) # type: ignore[assignment]
Review Comment:
So fast. 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] zhengruifeng commented on a diff in pull request #41670: [WIP][SPARK-43631][CONNECT][PS] Enable Series.interpolate with Spark Connect
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #41670:
URL: https://github.com/apache/spark/pull/41670#discussion_r1235198694
##########
python/pyspark/pandas/series.py:
##########
@@ -2257,27 +2257,44 @@ def _interpolate(
return self._psdf.copy()._psser_for(self._column_label)
scol = self.spark.column
- sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
- last_non_null = PySparkColumn(sql_utils.lastNonNull(scol._jc))
- null_index = PySparkColumn(sql_utils.nullIndex(scol._jc))
+ if is_remote():
+ from pyspark.sql.connect.functions import _invoke_function_over_columns
Review Comment:
will it minimize the changes if we put those functions into `python/pyspark/pandas/spark/functions.py`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] itholic commented on a diff in pull request #41670: [WIP][SPARK-43631][CONNECT][PS] Enable Series.interpolate with Spark Connect
Posted by "itholic (via GitHub)" <gi...@apache.org>.
itholic commented on code in PR #41670:
URL: https://github.com/apache/spark/pull/41670#discussion_r1234805869
##########
python/pyspark/pandas/series.py:
##########
@@ -2257,27 +2257,44 @@ def _interpolate(
return self._psdf.copy()._psser_for(self._column_label)
scol = self.spark.column
- sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
- last_non_null = PySparkColumn(sql_utils.lastNonNull(scol._jc))
- null_index = PySparkColumn(sql_utils.nullIndex(scol._jc))
+ if is_remote():
+ from pyspark.sql.connect.functions import _invoke_function_over_columns
+ last_non_null = _invoke_function_over_columns(
+ "last_non_null",
+ scol, # type: ignore[arg-type]
+ )
+ null_index = _invoke_function_over_columns(
+ "null_index",
+ scol, # type: ignore[arg-type]
+ )
+ else:
+ sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
+ last_non_null = PySparkColumn(
+ sql_utils.lastNonNull(scol._jc) # type: ignore[assignment]
+ )
+ null_index = PySparkColumn(sql_utils.nullIndex(scol._jc)) # type: ignore[assignment]
Review Comment:
Hi, @zhengruifeng Could you happen to help reviewing this PR when you find some time??
I made some modifications to remove JVM dependencies from `last_non_null` and `null_index` based on other similar PRs, but it still doesn't work properly in Spark Connect as follows:
```python
>>> import pyspark.pandas as ps
>>> import numpy as np
>>> psser = ps.Series([1, np.nan, 3], name="a")
>>> psser.interpolate()
/.../spark/python/pyspark/sql/connect/expressions.py:945: UserWarning: WARN WindowExpression: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.
warnings.warn(
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/.../spark/python/pyspark/pandas/series.py", line 2232, in interpolate
return self._interpolate(
File "/.../spark/python/pyspark/pandas/series.py", line 2372, in _interpolate
self._psdf._internal.with_new_spark_column(self._column_label, scol) # TODO: dtype?
File "/.../spark/python/pyspark/pandas/internal.py", line 1420, in with_new_spark_column
return self.with_new_columns(
File "/.../spark/python/pyspark/pandas/internal.py", line 1368, in with_new_columns
return self.copy(
File "/.../spark/python/pyspark/pandas/internal.py", line 1486, in copy
return InternalFrame(
File "/.../spark/python/pyspark/pandas/internal.py", line 743, in __init__
schema = spark_frame.select(data_spark_columns).schema
File "/.../spark/python/pyspark/sql/connect/dataframe.py", line 1665, in schema
return self._session.client.schema(query)
File "/.../spark/python/pyspark/sql/connect/client/core.py", line 783, in schema
schema = self._analyze(method="schema", plan=plan).schema
File "/.../spark/python/pyspark/sql/connect/client/core.py", line 964, in _analyze
self._handle_error(error)
File "/.../spark/python/pyspark/sql/connect/client/core.py", line 1212, in _handle_error
self._handle_rpc_error(error)
File "/.../spark/python/pyspark/sql/connect/client/core.py", line 1248, in _handle_rpc_error
raise convert_exception(info, status.message) from None
pyspark.errors.exceptions.connect.AnalysisException: Invalid call to dataType on unresolved object
```
Looking at the error message, it seems that somewhere the column is not being resolved correctly. Is there any possible cause that you can speculate about?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on pull request #41670: [SPARK-43631][CONNECT][PS] Enable Series.interpolate with Spark Connect
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on PR #41670:
URL: https://github.com/apache/spark/pull/41670#issuecomment-1610219148
merged to master
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] zhengruifeng commented on a diff in pull request #41670: [WIP][SPARK-43631][CONNECT][PS] Enable Series.interpolate with Spark Connect
Posted by "zhengruifeng (via GitHub)" <gi...@apache.org>.
zhengruifeng commented on code in PR #41670:
URL: https://github.com/apache/spark/pull/41670#discussion_r1234897887
##########
python/pyspark/pandas/series.py:
##########
@@ -2257,27 +2257,44 @@ def _interpolate(
return self._psdf.copy()._psser_for(self._column_label)
scol = self.spark.column
- sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
- last_non_null = PySparkColumn(sql_utils.lastNonNull(scol._jc))
- null_index = PySparkColumn(sql_utils.nullIndex(scol._jc))
+ if is_remote():
+ from pyspark.sql.connect.functions import _invoke_function_over_columns
+ last_non_null = _invoke_function_over_columns(
+ "last_non_null",
+ scol, # type: ignore[arg-type]
+ )
+ null_index = _invoke_function_over_columns(
+ "null_index",
+ scol, # type: ignore[arg-type]
+ )
+ else:
+ sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
+ last_non_null = PySparkColumn(
+ sql_utils.lastNonNull(scol._jc) # type: ignore[assignment]
+ )
+ null_index = PySparkColumn(sql_utils.nullIndex(scol._jc)) # type: ignore[assignment]
Review Comment:
should be resolved in https://github.com/apache/spark/pull/41672
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] itholic commented on a diff in pull request #41670: [WIP][SPARK-43631][CONNECT][PS] Enable Series.interpolate with Spark Connect
Posted by "itholic (via GitHub)" <gi...@apache.org>.
itholic commented on code in PR #41670:
URL: https://github.com/apache/spark/pull/41670#discussion_r1234805869
##########
python/pyspark/pandas/series.py:
##########
@@ -2257,27 +2257,44 @@ def _interpolate(
return self._psdf.copy()._psser_for(self._column_label)
scol = self.spark.column
- sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
- last_non_null = PySparkColumn(sql_utils.lastNonNull(scol._jc))
- null_index = PySparkColumn(sql_utils.nullIndex(scol._jc))
+ if is_remote():
+ from pyspark.sql.connect.functions import _invoke_function_over_columns
+ last_non_null = _invoke_function_over_columns(
+ "last_non_null",
+ scol, # type: ignore[arg-type]
+ )
+ null_index = _invoke_function_over_columns(
+ "null_index",
+ scol, # type: ignore[arg-type]
+ )
+ else:
+ sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
+ last_non_null = PySparkColumn(
+ sql_utils.lastNonNull(scol._jc) # type: ignore[assignment]
+ )
+ null_index = PySparkColumn(sql_utils.nullIndex(scol._jc)) # type: ignore[assignment]
Review Comment:
Hi, @zhengruifeng Could you happen to help reviewing this PR when you find some time??
I made some modifications to remove JVM dependencies from `last_non_null` and `null_index` based on other similar PRs, but it still doesn't work properly in Spark Connect as follows:
```python
>>> import pyspark.pandas as ps
>>> import numpy as np
>>> psser = ps.Series([1, np.nan, 3], name="a")
>>> psser.interpolate()
/.../spark/python/pyspark/sql/connect/expressions.py:945: UserWarning: WARN WindowExpression: No Partition Defined for Window operation! Moving all data to a single partition, this can cause serious performance degradation.
warnings.warn(
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/.../spark/python/pyspark/pandas/series.py", line 2232, in interpolate
return self._interpolate(
File "/.../spark/python/pyspark/pandas/series.py", line 2372, in _interpolate
self._psdf._internal.with_new_spark_column(self._column_label, scol) # TODO: dtype?
File "/.../spark/python/pyspark/pandas/internal.py", line 1420, in with_new_spark_column
return self.with_new_columns(
File "/.../spark/python/pyspark/pandas/internal.py", line 1368, in with_new_columns
return self.copy(
File "/.../spark/python/pyspark/pandas/internal.py", line 1486, in copy
return InternalFrame(
File "/.../spark/python/pyspark/pandas/internal.py", line 743, in __init__
schema = spark_frame.select(data_spark_columns).schema
File "/.../spark/python/pyspark/sql/connect/dataframe.py", line 1665, in schema
return self._session.client.schema(query)
File "/.../spark/python/pyspark/sql/connect/client/core.py", line 783, in schema
schema = self._analyze(method="schema", plan=plan).schema
File "/.../spark/python/pyspark/sql/connect/client/core.py", line 964, in _analyze
self._handle_error(error)
File "/.../spark/python/pyspark/sql/connect/client/core.py", line 1212, in _handle_error
self._handle_rpc_error(error)
File "/.../spark/python/pyspark/sql/connect/client/core.py", line 1248, in _handle_rpc_error
raise convert_exception(info, status.message) from None
pyspark.errors.exceptions.connect.AnalysisException: Invalid call to dataType on unresolved object
```
Looking at the error message, it seems that somewhere the column is not being resolved correctly. Is there any possible cause that you can speculate about?
This basically aims to support `interpolate` in Spark Connect, which is originally implemented by you IIRC, so I gently ask your help 🙏
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] itholic commented on a diff in pull request #41670: [WIP][SPARK-43631][CONNECT][PS] Enable Series.interpolate with Spark Connect
Posted by "itholic (via GitHub)" <gi...@apache.org>.
itholic commented on code in PR #41670:
URL: https://github.com/apache/spark/pull/41670#discussion_r1236271792
##########
python/pyspark/pandas/series.py:
##########
@@ -2257,27 +2257,44 @@ def _interpolate(
return self._psdf.copy()._psser_for(self._column_label)
scol = self.spark.column
- sql_utils = SparkContext._active_spark_context._jvm.PythonSQLUtils
- last_non_null = PySparkColumn(sql_utils.lastNonNull(scol._jc))
- null_index = PySparkColumn(sql_utils.nullIndex(scol._jc))
+ if is_remote():
+ from pyspark.sql.connect.functions import _invoke_function_over_columns
Review Comment:
Sounds good. Let me address the placement.
--
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