You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "harshmotw-db (via GitHub)" <gi...@apache.org> on 2024/11/06 01:47:41 UTC
[PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
harshmotw-db opened a new pull request, #48770:
URL: https://github.com/apache/spark/pull/48770
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'common/utils/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
This PR adds support for the Variant type in PySpark UDFs/UDTFs. Support is added in both modes - arrow and pickle - and support is also added in pandas UDFs.
### 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.
-->
After this change, users will be able to use the new Variant data type with UDFs, which is currently prohibited.
### 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, users should now be able to use Variants with Python UDFs.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Unit tests in all scenarios - arrow, pickle and pandas
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
No.
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1831618194
##########
sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala:
##########
@@ -146,19 +146,33 @@ private[sql] object ArrowUtils {
ArrowType.Struct.INSTANCE,
null,
Map("variant" -> "true").asJava)
+ val metadataFieldType = new FieldType(
+ false,
+ toArrowType(BinaryType, timeZoneId, largeVarTypes),
+ null,
+ Map("variant" -> "true").asJava
+ )
new Field(
name,
fieldType,
Seq(
toArrowField("value", BinaryType, false, timeZoneId, largeVarTypes),
- toArrowField("metadata", BinaryType, false, timeZoneId, largeVarTypes)).asJava)
+ new Field("metadata", metadataFieldType, Seq.empty[Field].asJava)).asJava)
case dataType =>
val fieldType =
new FieldType(nullable, toArrowType(dataType, timeZoneId, largeVarTypes), null)
new Field(name, fieldType, Seq.empty[Field].asJava)
}
}
+ def isVariantField(field: Field): Boolean = {
+ assert(field.getType.isInstanceOf[ArrowType.Struct])
Review Comment:
It should be an assert since the [callsite](https://github.com/apache/spark/blob/876d5cab7ffeb55b86516350af23aca8cae35afe/sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala#L187) already checks for struct.
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs/UDAFs [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #48770:
URL: https://github.com/apache/spark/pull/48770#issuecomment-2472052047
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1830289590
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/PythonUDF.scala:
##########
@@ -169,6 +148,23 @@ case class PythonUDAF(
override protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): PythonUDAF =
copy(children = newChildren)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
Review Comment:
@HyukjinKwon I haven't tested support in UDAFs yet, so I kept it disabled in this scenario. Can you point me to examples where Python UDAFs are tested?
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "ueshin (via GitHub)" <gi...@apache.org>.
ueshin commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1834889583
##########
python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py:
##########
@@ -752,46 +752,86 @@ def check_vectorized_udf_return_scalar(self):
def test_udf_with_variant_input(self):
df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
- from pyspark.sql.functions import col
- scalar_f = pandas_udf(lambda u: str(u), StringType())
+ scalar_f = pandas_udf(lambda u: u.apply(str), StringType(), PandasUDFType.SCALAR)
iter_f = pandas_udf(
- lambda it: map(lambda u: str(u), it), StringType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(str), it), StringType(), PandasUDFType.SCALAR_ITER
)
+ expected = [Row(udf="{0}".format(i)) for i in range(10)]
+
for f in [scalar_f, iter_f]:
- with self.assertRaises(AnalysisException) as ae:
- df.select(f(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={
- "sqlExpr": '"<lambda>(v)"',
- "dataType": "VARIANT",
- },
- )
+ result = df.select(f(col("v")).alias("udf")).collect()
+ self.assertEqual(result, expected)
def test_udf_with_variant_output(self):
- # Corresponds to a JSON string of {"a": "b"}.
- returned_variant = VariantVal(bytes([2, 1, 0, 0, 2, 5, 98]), bytes([1, 1, 0, 1, 97]))
- scalar_f = pandas_udf(lambda x: returned_variant, VariantType())
+ scalar_f = pandas_udf(
+ lambda u: u.apply(lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))), VariantType()
+ )
iter_f = pandas_udf(
- lambda it: map(lambda x: returned_variant, it), VariantType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(
+ lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))
Review Comment:
Sure, then shall we have comments on them about what these mean, and TODO comments with a JIRA ID?
```
# TODO(SPARK-?????): ...
```
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1831696110
##########
python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py:
##########
@@ -752,46 +752,87 @@ def check_vectorized_udf_return_scalar(self):
def test_udf_with_variant_input(self):
df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
- from pyspark.sql.functions import col
- scalar_f = pandas_udf(lambda u: str(u), StringType())
+ scalar_f = pandas_udf(lambda u: u.apply(str), StringType(), PandasUDFType.SCALAR)
Review Comment:
Yes, for the most part. I recall that for pandas UDFs to work, I also had to add changes in `arrow_to_pandas` and `_create_batch` too because they treat struct types in a special way. Example: https://github.com/apache/spark/pull/48770/files#r1831583273
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #48770:
URL: https://github.com/apache/spark/pull/48770#issuecomment-2466659766
Seems mostly fine to me
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "gene-db (via GitHub)" <gi...@apache.org>.
gene-db commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1831547607
##########
python/pyspark/sql/tests/test_udf.py:
##########
@@ -334,68 +334,72 @@ def test_udf_with_filter_function(self):
self.assertEqual(sel.collect(), [Row(key=1, value="1")])
def test_udf_with_variant_input(self):
- df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
-
- u = udf(lambda u: str(u), StringType())
- with self.assertRaises(AnalysisException) as ae:
- df.select(u(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={"sqlExpr": '"<lambda>(v)"', "dataType": "VARIANT"},
- )
+ for arrow_enabled in ["false", "true"]:
+ with self.sql_conf({"spark.sql.execution.pythonUDF.arrow.enabled": arrow_enabled}):
Review Comment:
If arrow is disabled, does it go through the pickle path?
##########
python/pyspark/sql/tests/test_udf.py:
##########
@@ -334,68 +334,72 @@ def test_udf_with_filter_function(self):
self.assertEqual(sel.collect(), [Row(key=1, value="1")])
def test_udf_with_variant_input(self):
- df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
-
- u = udf(lambda u: str(u), StringType())
- with self.assertRaises(AnalysisException) as ae:
- df.select(u(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={"sqlExpr": '"<lambda>(v)"', "dataType": "VARIANT"},
- )
+ for arrow_enabled in ["false", "true"]:
+ with self.sql_conf({"spark.sql.execution.pythonUDF.arrow.enabled": arrow_enabled}):
+ df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
+ u = udf(lambda u: str(u), StringType())
+ expected = [Row(udf="{0}".format(i)) for i in range(10)]
+ result = df.select(u(col("v")).alias("udf")).collect()
+ self.assertEqual(result, expected)
def test_udf_with_complex_variant_input(self):
- df = self.spark.range(0, 10).selectExpr(
- "named_struct('v', parse_json(cast(id as string))) struct_of_v"
- )
-
- u = udf(lambda u: str(u), StringType())
-
- with self.assertRaises(AnalysisException) as ae:
- df.select(u(col("struct_of_v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={
- "sqlExpr": '"<lambda>(struct_of_v)"',
- "dataType": "STRUCT<v: VARIANT NOT NULL>",
- },
- )
+ for arrow_enabled in ["false", "true"]:
+ with self.sql_conf({"spark.sql.execution.pythonUDF.arrow.enabled": arrow_enabled}):
+ df = self.spark.range(0, 10).selectExpr(
+ "named_struct('v', parse_json(cast(id as string))) struct_of_v"
+ )
+ u = udf(lambda u: str(u["v"]), StringType())
+ result = df.select(u(col("struct_of_v"))).collect()
+ expected = [Row(udf="{0}".format(i)) for i in range(10)]
+ self.assertEqual(result, expected)
def test_udf_with_variant_output(self):
- # The variant value returned corresponds to a JSON string of {"a": "b"}.
- u = udf(
- lambda: VariantVal(bytes([2, 1, 0, 0, 2, 5, 98]), bytes([1, 1, 0, 1, 97])),
- VariantType(),
- )
-
- with self.assertRaises(AnalysisException) as ae:
- self.spark.range(0, 10).select(u()).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_OUTPUT_TYPE",
- messageParameters={"sqlExpr": '"<lambda>()"', "dataType": "VARIANT"},
- )
+ for arrow_enabled in ["false", "true"]:
+ with self.sql_conf({"spark.sql.execution.pythonUDF.arrow.enabled": arrow_enabled}):
+ # The variant value returned corresponds to a JSON string of {"a": "<a-j>"}.
+ u = udf(
+ lambda i: VariantVal(
+ bytes([2, 1, 0, 0, 2, 5, 97 + i]), bytes([1, 1, 0, 1, 97])
+ ),
+ VariantType(),
+ )
+ result = self.spark.range(0, 10).select(
+ u(col("id")).cast("string").alias("udf")
+ ).collect()
+ expected = [Row(udf=f"{{\"a\":\"{chr(97 + i)}\"}}") for i in range(10)]
+ self.assertEqual(result, expected)
def test_udf_with_complex_variant_output(self):
- # The variant value returned corresponds to a JSON string of {"a": "b"}.
- u = udf(
- lambda: {"v", VariantVal(bytes([2, 1, 0, 0, 2, 5, 98]), bytes([1, 1, 0, 1, 97]))},
- MapType(StringType(), VariantType()),
- )
-
- with self.assertRaises(AnalysisException) as ae:
- self.spark.range(0, 10).select(u()).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_OUTPUT_TYPE",
- messageParameters={"sqlExpr": '"<lambda>()"', "dataType": "MAP<STRING, VARIANT>"},
- )
+ for arrow_enabled in ["false", "true"]:
+ with self.sql_conf({"spark.sql.execution.pythonUDF.arrow.enabled": arrow_enabled}):
+ # The variant value returned corresponds to a JSON string of {"a": "<a-j>"}.
+ u = udf(
+ lambda i: {
+ "v": VariantVal(bytes([2, 1, 0, 0, 2, 5, 97 + i]), bytes([1, 1, 0, 1, 97]))
+ },
+ MapType(StringType(), VariantType()),
+ )
+ result = self.spark.range(0, 10).select(
+ u(col("id")).cast("string").alias("udf")
+ ).collect()
+ expected = [Row(udf=f"{{v -> {{\"a\":\"{chr(97 + i)}\"}}}}") for i in range(10)]
+ self.assertEqual(result, expected)
+
+ def test_chained_udfs_with_variant(self):
Review Comment:
can we have another one with changed udfs, with a nested variant, like `array<variant>` or `struct<int, variant>` or something?
##########
python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py:
##########
@@ -752,46 +752,87 @@ def check_vectorized_udf_return_scalar(self):
def test_udf_with_variant_input(self):
df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
- from pyspark.sql.functions import col
- scalar_f = pandas_udf(lambda u: str(u), StringType())
+ scalar_f = pandas_udf(lambda u: u.apply(str), StringType(), PandasUDFType.SCALAR)
iter_f = pandas_udf(
- lambda it: map(lambda u: str(u), it), StringType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(str), it), StringType(), PandasUDFType.SCALAR_ITER
)
+ expected = [Row(udf="{0}".format(i)) for i in range(10)]
+
for f in [scalar_f, iter_f]:
- with self.assertRaises(AnalysisException) as ae:
- df.select(f(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={
- "sqlExpr": '"<lambda>(v)"',
- "dataType": "VARIANT",
- },
- )
+ result = df.select(f(col("v")).alias("udf")).collect()
+ self.assertEqual(result, expected)
def test_udf_with_variant_output(self):
- # Corresponds to a JSON string of {"a": "b"}.
- returned_variant = VariantVal(bytes([2, 1, 0, 0, 2, 5, 98]), bytes([1, 1, 0, 1, 97]))
- scalar_f = pandas_udf(lambda x: returned_variant, VariantType())
+ scalar_f = pandas_udf(
+ lambda u: u.apply(lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))), VariantType()
+ )
iter_f = pandas_udf(
- lambda it: map(lambda x: returned_variant, it), VariantType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(
+ lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))
+ ), it),
+ VariantType(),
+ PandasUDFType.SCALAR_ITER
)
+ expected = [Row(udf=i) for i in range(10)]
Review Comment:
Do we have a `pandas_udf` test case with a nested variant output?
##########
python/pyspark/sql/pandas/types.py:
##########
@@ -1295,6 +1306,16 @@ def convert_udt(value: Any) -> Any:
return convert_udt
+ elif isinstance(dt, VariantType):
+ def convert_variant(variant: Any) -> Any:
Review Comment:
What is this conversion function supposed to return? It is expecting a python `VariantVal`, and then returns a dict?
##########
python/pyspark/sql/pandas/serializers.py:
##########
@@ -420,7 +421,11 @@ def __init__(
def arrow_to_pandas(self, arrow_column):
import pyarrow.types as types
- if self._df_for_struct and types.is_struct(arrow_column.type):
+ if (
+ self._df_for_struct
+ and types.is_struct(arrow_column.type)
+ and not is_variant(arrow_column.type)
+ ):
import pandas as pd
Review Comment:
Why do we want to avoid this path for variant types? Maybe we should leave a comment explaining it.
##########
python/pyspark/sql/pandas/types.py:
##########
@@ -221,6 +221,15 @@ def to_arrow_schema(
return pa.schema(fields)
+def is_variant(at: "pa.DataType") -> bool:
+ """Check if a PyArrow struct data type represents a variant"""
+ import pyarrow.types as types
+ assert types.is_struct(at)
Review Comment:
Why is this an assert? Should this just return false if it is not a struct?
##########
sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala:
##########
@@ -146,19 +146,33 @@ private[sql] object ArrowUtils {
ArrowType.Struct.INSTANCE,
null,
Map("variant" -> "true").asJava)
+ val metadataFieldType = new FieldType(
+ false,
+ toArrowType(BinaryType, timeZoneId, largeVarTypes),
+ null,
+ Map("variant" -> "true").asJava
+ )
new Field(
name,
fieldType,
Seq(
toArrowField("value", BinaryType, false, timeZoneId, largeVarTypes),
- toArrowField("metadata", BinaryType, false, timeZoneId, largeVarTypes)).asJava)
+ new Field("metadata", metadataFieldType, Seq.empty[Field].asJava)).asJava)
case dataType =>
val fieldType =
new FieldType(nullable, toArrowType(dataType, timeZoneId, largeVarTypes), null)
new Field(name, fieldType, Seq.empty[Field].asJava)
}
}
+ def isVariantField(field: Field): Boolean = {
+ assert(field.getType.isInstanceOf[ArrowType.Struct])
Review Comment:
Should this assert, or should it just return false when it is not a struct?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/PythonUDF.scala:
##########
@@ -169,6 +148,23 @@ case class PythonUDAF(
override protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): PythonUDAF =
copy(children = newChildren)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val check = super.checkInputDataTypes()
+ if (check.isFailure) {
+ check
+ } else {
+ val exprReturningVariant = children.collectFirst {
+ case e: Expression if VariantExpressionEvalUtils.typeContainsVariant(e.dataType) => e
+ }
+ exprReturningVariant match {
+ case Some(e) => TypeCheckResult.DataTypeMismatch(
+ errorSubClass = "UNSUPPORTED_UDF_INPUT_TYPE",
Review Comment:
Does UDAF also use the error class for UDF?
##########
python/pyspark/sql/tests/test_udtf.py:
##########
@@ -2530,6 +2531,28 @@ def terminate(self):
[Row(current=4, total=4), Row(current=13, total=4), Row(current=20, total=1)],
)
+ def test_udtf_with_variant_input(self):
Review Comment:
Are we able to get variations of these 2 tests, with nested variant types?
##########
python/pyspark/sql/pandas/types.py:
##########
@@ -171,7 +171,7 @@ def to_arrow_type(
elif type(dt) == VariantType:
fields = [
pa.field("value", pa.binary(), nullable=False),
- pa.field("metadata", pa.binary(), nullable=False),
+ pa.field("metadata", pa.binary(), nullable=False, metadata={b"variant": b"true"}),
Review Comment:
Should we comment on the scheme we are using to "tag" an arrow struct as a variant?
It looks like we are attaching some metadata to the `metadata` field of the struct?
##########
python/pyspark/sql/tests/test_udf.py:
##########
@@ -334,68 +334,72 @@ def test_udf_with_filter_function(self):
self.assertEqual(sel.collect(), [Row(key=1, value="1")])
def test_udf_with_variant_input(self):
- df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
-
- u = udf(lambda u: str(u), StringType())
- with self.assertRaises(AnalysisException) as ae:
- df.select(u(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={"sqlExpr": '"<lambda>(v)"', "dataType": "VARIANT"},
- )
+ for arrow_enabled in ["false", "true"]:
+ with self.sql_conf({"spark.sql.execution.pythonUDF.arrow.enabled": arrow_enabled}):
+ df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
+ u = udf(lambda u: str(u), StringType())
+ expected = [Row(udf="{0}".format(i)) for i in range(10)]
+ result = df.select(u(col("v")).alias("udf")).collect()
+ self.assertEqual(result, expected)
def test_udf_with_complex_variant_input(self):
- df = self.spark.range(0, 10).selectExpr(
- "named_struct('v', parse_json(cast(id as string))) struct_of_v"
- )
-
- u = udf(lambda u: str(u), StringType())
-
- with self.assertRaises(AnalysisException) as ae:
- df.select(u(col("struct_of_v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={
- "sqlExpr": '"<lambda>(struct_of_v)"',
- "dataType": "STRUCT<v: VARIANT NOT NULL>",
- },
- )
+ for arrow_enabled in ["false", "true"]:
+ with self.sql_conf({"spark.sql.execution.pythonUDF.arrow.enabled": arrow_enabled}):
+ df = self.spark.range(0, 10).selectExpr(
+ "named_struct('v', parse_json(cast(id as string))) struct_of_v"
+ )
+ u = udf(lambda u: str(u["v"]), StringType())
+ result = df.select(u(col("struct_of_v"))).collect()
Review Comment:
can we also add unit tests with inputs of nested types
- `array<variant>`
- `map<string, variant>`
##########
python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py:
##########
@@ -752,46 +752,87 @@ def check_vectorized_udf_return_scalar(self):
def test_udf_with_variant_input(self):
df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
Review Comment:
Do we have a `pandas_udf` test case with a nested variant input?
##########
python/pyspark/sql/pandas/types.py:
##########
@@ -221,6 +221,15 @@ def to_arrow_schema(
return pa.schema(fields)
+def is_variant(at: "pa.DataType") -> bool:
+ """Check if a PyArrow struct data type represents a variant"""
+ import pyarrow.types as types
+ assert types.is_struct(at)
+
+ return any((field.name == "metadata" and b"variant" in field.metadata and
Review Comment:
Shouldn't we check that the fields are `metadata` and `value`?
##########
python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py:
##########
@@ -752,46 +752,87 @@ def check_vectorized_udf_return_scalar(self):
def test_udf_with_variant_input(self):
df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
- from pyspark.sql.functions import col
- scalar_f = pandas_udf(lambda u: str(u), StringType())
+ scalar_f = pandas_udf(lambda u: u.apply(str), StringType(), PandasUDFType.SCALAR)
iter_f = pandas_udf(
- lambda it: map(lambda u: str(u), it), StringType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(str), it), StringType(), PandasUDFType.SCALAR_ITER
)
+ expected = [Row(udf="{0}".format(i)) for i in range(10)]
+
for f in [scalar_f, iter_f]:
- with self.assertRaises(AnalysisException) as ae:
- df.select(f(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={
- "sqlExpr": '"<lambda>(v)"',
- "dataType": "VARIANT",
- },
- )
+ result = df.select(f(col("v")).alias("udf")).collect()
+ self.assertEqual(result, expected)
def test_udf_with_variant_output(self):
- # Corresponds to a JSON string of {"a": "b"}.
- returned_variant = VariantVal(bytes([2, 1, 0, 0, 2, 5, 98]), bytes([1, 1, 0, 1, 97]))
- scalar_f = pandas_udf(lambda x: returned_variant, VariantType())
+ scalar_f = pandas_udf(
+ lambda u: u.apply(lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))), VariantType()
+ )
iter_f = pandas_udf(
- lambda it: map(lambda x: returned_variant, it), VariantType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(
+ lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))
+ ), it),
+ VariantType(),
+ PandasUDFType.SCALAR_ITER
)
+ expected = [Row(udf=i) for i in range(10)]
+
for f in [scalar_f, iter_f]:
- with self.assertRaises(AnalysisException) as ae:
- self.spark.range(0, 10).select(f()).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_OUTPUT_TYPE",
- messageParameters={
- "sqlExpr": '"<lambda>()"',
- "dataType": "VARIANT",
- },
- )
+ # with self.assertRaises(AnalysisException) as ae:
Review Comment:
remove?
##########
sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala:
##########
@@ -146,19 +146,33 @@ private[sql] object ArrowUtils {
ArrowType.Struct.INSTANCE,
null,
Map("variant" -> "true").asJava)
+ val metadataFieldType = new FieldType(
Review Comment:
Do we have 2 ways of tagging something a variant? The line above is some metadata, but we also have this other metadata thing? This is confusing, so could you explain this more with some comments? Do we need both schemes?
##########
python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py:
##########
@@ -752,46 +752,87 @@ def check_vectorized_udf_return_scalar(self):
def test_udf_with_variant_input(self):
df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
- from pyspark.sql.functions import col
- scalar_f = pandas_udf(lambda u: str(u), StringType())
+ scalar_f = pandas_udf(lambda u: u.apply(str), StringType(), PandasUDFType.SCALAR)
Review Comment:
Does `pandas_udf` go through the same path as an arrow udf path?
##########
python/pyspark/sql/pandas/serializers.py:
##########
@@ -505,7 +510,12 @@ def _create_batch(self, series):
arrs = []
for s, t in series:
- if self._struct_in_pandas == "dict" and t is not None and pa.types.is_struct(t):
+ if (
+ self._struct_in_pandas == "dict"
+ and t is not None
+ and pa.types.is_struct(t)
+ and not is_variant(t)
+ ):
# A pandas UDF should return pd.DataFrame when the return type is a struct type.
Review Comment:
Why do we want to avoid this path for variant types? Maybe we should leave a comment explaining it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on PR #48770:
URL: https://github.com/apache/spark/pull/48770#issuecomment-2469349966
@HyukjinKwon @ueshin I'm not sure why SparkConnectSessionHolderSuite is failing on this PR. I am not able to repro the failure on local. Can you look into 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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "ueshin (via GitHub)" <gi...@apache.org>.
ueshin commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1834889583
##########
python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py:
##########
@@ -752,46 +752,86 @@ def check_vectorized_udf_return_scalar(self):
def test_udf_with_variant_input(self):
df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
- from pyspark.sql.functions import col
- scalar_f = pandas_udf(lambda u: str(u), StringType())
+ scalar_f = pandas_udf(lambda u: u.apply(str), StringType(), PandasUDFType.SCALAR)
iter_f = pandas_udf(
- lambda it: map(lambda u: str(u), it), StringType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(str), it), StringType(), PandasUDFType.SCALAR_ITER
)
+ expected = [Row(udf="{0}".format(i)) for i in range(10)]
+
for f in [scalar_f, iter_f]:
- with self.assertRaises(AnalysisException) as ae:
- df.select(f(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={
- "sqlExpr": '"<lambda>(v)"',
- "dataType": "VARIANT",
- },
- )
+ result = df.select(f(col("v")).alias("udf")).collect()
+ self.assertEqual(result, expected)
def test_udf_with_variant_output(self):
- # Corresponds to a JSON string of {"a": "b"}.
- returned_variant = VariantVal(bytes([2, 1, 0, 0, 2, 5, 98]), bytes([1, 1, 0, 1, 97]))
- scalar_f = pandas_udf(lambda x: returned_variant, VariantType())
+ scalar_f = pandas_udf(
+ lambda u: u.apply(lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))), VariantType()
+ )
iter_f = pandas_udf(
- lambda it: map(lambda x: returned_variant, it), VariantType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(
+ lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))
Review Comment:
Sure, then shall we have comments on them about what these mean, and TODO comments with a JIRA ID to track the easy API implementation?
```
# TODO(SPARK-?????): ...
```
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "ueshin (via GitHub)" <gi...@apache.org>.
ueshin commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1831910267
##########
python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py:
##########
@@ -752,46 +752,86 @@ def check_vectorized_udf_return_scalar(self):
def test_udf_with_variant_input(self):
df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
- from pyspark.sql.functions import col
- scalar_f = pandas_udf(lambda u: str(u), StringType())
+ scalar_f = pandas_udf(lambda u: u.apply(str), StringType(), PandasUDFType.SCALAR)
iter_f = pandas_udf(
- lambda it: map(lambda u: str(u), it), StringType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(str), it), StringType(), PandasUDFType.SCALAR_ITER
)
+ expected = [Row(udf="{0}".format(i)) for i in range(10)]
+
for f in [scalar_f, iter_f]:
- with self.assertRaises(AnalysisException) as ae:
- df.select(f(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={
- "sqlExpr": '"<lambda>(v)"',
- "dataType": "VARIANT",
- },
- )
+ result = df.select(f(col("v")).alias("udf")).collect()
+ self.assertEqual(result, expected)
def test_udf_with_variant_output(self):
- # Corresponds to a JSON string of {"a": "b"}.
- returned_variant = VariantVal(bytes([2, 1, 0, 0, 2, 5, 98]), bytes([1, 1, 0, 1, 97]))
- scalar_f = pandas_udf(lambda x: returned_variant, VariantType())
+ scalar_f = pandas_udf(
+ lambda u: u.apply(lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))), VariantType()
+ )
iter_f = pandas_udf(
- lambda it: map(lambda x: returned_variant, it), VariantType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(
+ lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))
Review Comment:
I guess we need something like `VariantVal.fromPython()` and `VariantType.fromJson()`?
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1831626706
##########
python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py:
##########
@@ -752,46 +752,87 @@ def check_vectorized_udf_return_scalar(self):
def test_udf_with_variant_input(self):
df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
- from pyspark.sql.functions import col
- scalar_f = pandas_udf(lambda u: str(u), StringType())
+ scalar_f = pandas_udf(lambda u: u.apply(str), StringType(), PandasUDFType.SCALAR)
iter_f = pandas_udf(
- lambda it: map(lambda u: str(u), it), StringType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(str), it), StringType(), PandasUDFType.SCALAR_ITER
)
+ expected = [Row(udf="{0}".format(i)) for i in range(10)]
+
for f in [scalar_f, iter_f]:
- with self.assertRaises(AnalysisException) as ae:
- df.select(f(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={
- "sqlExpr": '"<lambda>(v)"',
- "dataType": "VARIANT",
- },
- )
+ result = df.select(f(col("v")).alias("udf")).collect()
+ self.assertEqual(result, expected)
def test_udf_with_variant_output(self):
- # Corresponds to a JSON string of {"a": "b"}.
- returned_variant = VariantVal(bytes([2, 1, 0, 0, 2, 5, 98]), bytes([1, 1, 0, 1, 97]))
- scalar_f = pandas_udf(lambda x: returned_variant, VariantType())
+ scalar_f = pandas_udf(
+ lambda u: u.apply(lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))), VariantType()
+ )
iter_f = pandas_udf(
- lambda it: map(lambda x: returned_variant, it), VariantType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(
+ lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))
+ ), it),
+ VariantType(),
+ PandasUDFType.SCALAR_ITER
)
+ expected = [Row(udf=i) for i in range(10)]
Review Comment:
I did add a [nested variant test with chained pandas UDFs](https://github.com/apache/spark/blob/876d5cab7ffeb55b86516350af23aca8cae35afe/python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py#L809) but I should also add tests for isolated UDFs.
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1831624144
##########
python/pyspark/sql/tests/test_udf.py:
##########
@@ -334,68 +334,72 @@ def test_udf_with_filter_function(self):
self.assertEqual(sel.collect(), [Row(key=1, value="1")])
def test_udf_with_variant_input(self):
- df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
-
- u = udf(lambda u: str(u), StringType())
- with self.assertRaises(AnalysisException) as ae:
- df.select(u(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={"sqlExpr": '"<lambda>(v)"', "dataType": "VARIANT"},
- )
+ for arrow_enabled in ["false", "true"]:
+ with self.sql_conf({"spark.sql.execution.pythonUDF.arrow.enabled": arrow_enabled}):
Review Comment:
Yes
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "ueshin (via GitHub)" <gi...@apache.org>.
ueshin commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1831904054
##########
python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py:
##########
@@ -752,46 +752,86 @@ def check_vectorized_udf_return_scalar(self):
def test_udf_with_variant_input(self):
df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
- from pyspark.sql.functions import col
- scalar_f = pandas_udf(lambda u: str(u), StringType())
+ scalar_f = pandas_udf(lambda u: u.apply(str), StringType(), PandasUDFType.SCALAR)
iter_f = pandas_udf(
- lambda it: map(lambda u: str(u), it), StringType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(str), it), StringType(), PandasUDFType.SCALAR_ITER
)
+ expected = [Row(udf="{0}".format(i)) for i in range(10)]
+
for f in [scalar_f, iter_f]:
- with self.assertRaises(AnalysisException) as ae:
- df.select(f(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={
- "sqlExpr": '"<lambda>(v)"',
- "dataType": "VARIANT",
- },
- )
+ result = df.select(f(col("v")).alias("udf")).collect()
+ self.assertEqual(result, expected)
def test_udf_with_variant_output(self):
- # Corresponds to a JSON string of {"a": "b"}.
- returned_variant = VariantVal(bytes([2, 1, 0, 0, 2, 5, 98]), bytes([1, 1, 0, 1, 97]))
- scalar_f = pandas_udf(lambda x: returned_variant, VariantType())
+ scalar_f = pandas_udf(
+ lambda u: u.apply(lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))), VariantType()
+ )
iter_f = pandas_udf(
- lambda it: map(lambda x: returned_variant, it), VariantType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(
+ lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))
Review Comment:
What does `VariantVal(bytes([12, i]), bytes([1, 0, 0]))` mean?
How can we properly create `VariantVal` as a user?
##########
python/pyspark/sql/tests/test_udf.py:
##########
@@ -334,68 +334,118 @@ def test_udf_with_filter_function(self):
self.assertEqual(sel.collect(), [Row(key=1, value="1")])
def test_udf_with_variant_input(self):
- df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
-
- u = udf(lambda u: str(u), StringType())
- with self.assertRaises(AnalysisException) as ae:
- df.select(u(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={"sqlExpr": '"<lambda>(v)"', "dataType": "VARIANT"},
- )
+ for arrow_enabled in ["false", "true"]:
+ with self.sql_conf({"spark.sql.execution.pythonUDF.arrow.enabled": arrow_enabled}):
Review Comment:
We don't need this here. There is `PythonUDFArrowTests` sharing tests in `BaseUDFTestsMixin` with the config enabled.
##########
python/pyspark/sql/tests/test_udf.py:
##########
@@ -334,68 +334,118 @@ def test_udf_with_filter_function(self):
self.assertEqual(sel.collect(), [Row(key=1, value="1")])
def test_udf_with_variant_input(self):
- df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
-
- u = udf(lambda u: str(u), StringType())
- with self.assertRaises(AnalysisException) as ae:
- df.select(u(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={"sqlExpr": '"<lambda>(v)"', "dataType": "VARIANT"},
- )
+ for arrow_enabled in ["false", "true"]:
+ with self.sql_conf({"spark.sql.execution.pythonUDF.arrow.enabled": arrow_enabled}):
+ df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
+ u = udf(lambda u: str(u), StringType())
+ expected = [Row(udf="{0}".format(i)) for i in range(10)]
+ result = df.select(u(col("v")).alias("udf")).collect()
+ self.assertEqual(result, expected)
def test_udf_with_complex_variant_input(self):
- df = self.spark.range(0, 10).selectExpr(
- "named_struct('v', parse_json(cast(id as string))) struct_of_v"
- )
-
- u = udf(lambda u: str(u), StringType())
-
- with self.assertRaises(AnalysisException) as ae:
- df.select(u(col("struct_of_v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={
- "sqlExpr": '"<lambda>(struct_of_v)"',
- "dataType": "STRUCT<v: VARIANT NOT NULL>",
- },
- )
+ for arrow_enabled in ["false", "true"]:
+ with self.sql_conf({"spark.sql.execution.pythonUDF.arrow.enabled": arrow_enabled}):
Review Comment:
ditto, and following tests in `BaseUDFTestsMixin`.
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1834847538
##########
python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py:
##########
@@ -752,46 +752,86 @@ def check_vectorized_udf_return_scalar(self):
def test_udf_with_variant_input(self):
df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
- from pyspark.sql.functions import col
- scalar_f = pandas_udf(lambda u: str(u), StringType())
+ scalar_f = pandas_udf(lambda u: u.apply(str), StringType(), PandasUDFType.SCALAR)
iter_f = pandas_udf(
- lambda it: map(lambda u: str(u), it), StringType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(str), it), StringType(), PandasUDFType.SCALAR_ITER
)
+ expected = [Row(udf="{0}".format(i)) for i in range(10)]
+
for f in [scalar_f, iter_f]:
- with self.assertRaises(AnalysisException) as ae:
- df.select(f(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={
- "sqlExpr": '"<lambda>(v)"',
- "dataType": "VARIANT",
- },
- )
+ result = df.select(f(col("v")).alias("udf")).collect()
+ self.assertEqual(result, expected)
def test_udf_with_variant_output(self):
- # Corresponds to a JSON string of {"a": "b"}.
- returned_variant = VariantVal(bytes([2, 1, 0, 0, 2, 5, 98]), bytes([1, 1, 0, 1, 97]))
- scalar_f = pandas_udf(lambda x: returned_variant, VariantType())
+ scalar_f = pandas_udf(
+ lambda u: u.apply(lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))), VariantType()
+ )
iter_f = pandas_udf(
- lambda it: map(lambda x: returned_variant, it), VariantType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(
+ lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))
Review Comment:
`VariantVal(bytes([12, i]), bytes([1, 0, 0]))` is a variant representing the int8 value `i`.
About your second point, currently there is no easy API to construct a Variant in Python. Something simple like a `parse_json` API should be implemented in the future but that is a non-trivial change. This API just deals with the communication between the JVM and the Python worker using Pickle and Arrow.
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on PR #48770:
URL: https://github.com/apache/spark/pull/48770#issuecomment-2471368765
Noting that the tests that were failing earlier passed on the latest commit. Seems like a broken test on old versions
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs/UDAFs [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #48770: [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs/UDAFs
URL: https://github.com/apache/spark/pull/48770
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1831611328
##########
python/pyspark/sql/pandas/types.py:
##########
@@ -221,6 +221,15 @@ def to_arrow_schema(
return pa.schema(fields)
+def is_variant(at: "pa.DataType") -> bool:
+ """Check if a PyArrow struct data type represents a variant"""
+ import pyarrow.types as types
+ assert types.is_struct(at)
Review Comment:
This function is intended to be called with `at.is_struct` so I want to prevent developers from using this function with non-struct types. I should add a comment.
Checking `is_struct` in this function adds cost in production (where I'm assuming Python runs in optimized mode so asserts are disabled).
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1834847538
##########
python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py:
##########
@@ -752,46 +752,86 @@ def check_vectorized_udf_return_scalar(self):
def test_udf_with_variant_input(self):
df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
- from pyspark.sql.functions import col
- scalar_f = pandas_udf(lambda u: str(u), StringType())
+ scalar_f = pandas_udf(lambda u: u.apply(str), StringType(), PandasUDFType.SCALAR)
iter_f = pandas_udf(
- lambda it: map(lambda u: str(u), it), StringType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(str), it), StringType(), PandasUDFType.SCALAR_ITER
)
+ expected = [Row(udf="{0}".format(i)) for i in range(10)]
+
for f in [scalar_f, iter_f]:
- with self.assertRaises(AnalysisException) as ae:
- df.select(f(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={
- "sqlExpr": '"<lambda>(v)"',
- "dataType": "VARIANT",
- },
- )
+ result = df.select(f(col("v")).alias("udf")).collect()
+ self.assertEqual(result, expected)
def test_udf_with_variant_output(self):
- # Corresponds to a JSON string of {"a": "b"}.
- returned_variant = VariantVal(bytes([2, 1, 0, 0, 2, 5, 98]), bytes([1, 1, 0, 1, 97]))
- scalar_f = pandas_udf(lambda x: returned_variant, VariantType())
+ scalar_f = pandas_udf(
+ lambda u: u.apply(lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))), VariantType()
+ )
iter_f = pandas_udf(
- lambda it: map(lambda x: returned_variant, it), VariantType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(
+ lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))
Review Comment:
`VariantVal(bytes([12, i]), bytes([1, 0, 0]))` is a variant representing the int8 value `i`.
About your second point, currently there is no easy API to construct a Variant in Python. Something simple like a `parse_json` API should be implemented in the future but that is a non-trivial change. This API just deals with the communication between the JVM and the Python worker.
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1835636736
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/PythonUDF.scala:
##########
@@ -169,6 +148,23 @@ case class PythonUDAF(
override protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): PythonUDAF =
copy(children = newChildren)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
Review Comment:
```python
import pandas as pd
from pyspark.sql.functions import pandas_udf
df = spark.createDataFrame(
[(1, 1.0), (1, 2.0), (2, 3.0), (2, 5.0), (2, 10.0)],
("id", "v"))
# Declare the function and create the UDF
@pandas_udf("double")
def mean_udf(v: pd.Series) -> float:
return v.mean()
df.select(mean_udf(df['v'])).show()
```
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1834847538
##########
python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py:
##########
@@ -752,46 +752,86 @@ def check_vectorized_udf_return_scalar(self):
def test_udf_with_variant_input(self):
df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
- from pyspark.sql.functions import col
- scalar_f = pandas_udf(lambda u: str(u), StringType())
+ scalar_f = pandas_udf(lambda u: u.apply(str), StringType(), PandasUDFType.SCALAR)
iter_f = pandas_udf(
- lambda it: map(lambda u: str(u), it), StringType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(str), it), StringType(), PandasUDFType.SCALAR_ITER
)
+ expected = [Row(udf="{0}".format(i)) for i in range(10)]
+
for f in [scalar_f, iter_f]:
- with self.assertRaises(AnalysisException) as ae:
- df.select(f(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={
- "sqlExpr": '"<lambda>(v)"',
- "dataType": "VARIANT",
- },
- )
+ result = df.select(f(col("v")).alias("udf")).collect()
+ self.assertEqual(result, expected)
def test_udf_with_variant_output(self):
- # Corresponds to a JSON string of {"a": "b"}.
- returned_variant = VariantVal(bytes([2, 1, 0, 0, 2, 5, 98]), bytes([1, 1, 0, 1, 97]))
- scalar_f = pandas_udf(lambda x: returned_variant, VariantType())
+ scalar_f = pandas_udf(
+ lambda u: u.apply(lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))), VariantType()
+ )
iter_f = pandas_udf(
- lambda it: map(lambda x: returned_variant, it), VariantType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(
+ lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))
Review Comment:
`VariantVal(bytes([12, i]), bytes([1, 0, 0]))` is a variant representing the int8 value `i`.
About your second point, currently there is no easy API to construct a Variant in Python. Something simple like a `parse_json` API should be implemented in the future but that is a non-trivial change. This PR just deals with the communication between the JVM and the Python worker using Pickle and Arrow.
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "ueshin (via GitHub)" <gi...@apache.org>.
ueshin commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1834889583
##########
python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py:
##########
@@ -752,46 +752,86 @@ def check_vectorized_udf_return_scalar(self):
def test_udf_with_variant_input(self):
df = self.spark.range(0, 10).selectExpr("parse_json(cast(id as string)) v")
- from pyspark.sql.functions import col
- scalar_f = pandas_udf(lambda u: str(u), StringType())
+ scalar_f = pandas_udf(lambda u: u.apply(str), StringType(), PandasUDFType.SCALAR)
iter_f = pandas_udf(
- lambda it: map(lambda u: str(u), it), StringType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(str), it), StringType(), PandasUDFType.SCALAR_ITER
)
+ expected = [Row(udf="{0}".format(i)) for i in range(10)]
+
for f in [scalar_f, iter_f]:
- with self.assertRaises(AnalysisException) as ae:
- df.select(f(col("v"))).collect()
-
- self.check_error(
- exception=ae.exception,
- errorClass="DATATYPE_MISMATCH.UNSUPPORTED_UDF_INPUT_TYPE",
- messageParameters={
- "sqlExpr": '"<lambda>(v)"',
- "dataType": "VARIANT",
- },
- )
+ result = df.select(f(col("v")).alias("udf")).collect()
+ self.assertEqual(result, expected)
def test_udf_with_variant_output(self):
- # Corresponds to a JSON string of {"a": "b"}.
- returned_variant = VariantVal(bytes([2, 1, 0, 0, 2, 5, 98]), bytes([1, 1, 0, 1, 97]))
- scalar_f = pandas_udf(lambda x: returned_variant, VariantType())
+ scalar_f = pandas_udf(
+ lambda u: u.apply(lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))), VariantType()
+ )
iter_f = pandas_udf(
- lambda it: map(lambda x: returned_variant, it), VariantType(), PandasUDFType.SCALAR_ITER
+ lambda it: map(lambda u: u.apply(
+ lambda i: VariantVal(bytes([12, i]), bytes([1, 0, 0]))
Review Comment:
Sure, then shall we have comments on them about what this means, and TODO comment with a JIRA ID?
```
# TODO(SPARK-?????): ...
```
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1831622208
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/PythonUDF.scala:
##########
@@ -169,6 +148,23 @@ case class PythonUDAF(
override protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): PythonUDAF =
copy(children = newChildren)
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ val check = super.checkInputDataTypes()
+ if (check.isFailure) {
+ check
+ } else {
+ val exprReturningVariant = children.collectFirst {
+ case e: Expression if VariantExpressionEvalUtils.typeContainsVariant(e.dataType) => e
+ }
+ exprReturningVariant match {
+ case Some(e) => TypeCheckResult.DataTypeMismatch(
+ errorSubClass = "UNSUPPORTED_UDF_INPUT_TYPE",
Review Comment:
It was using the same error class before this PR as PythonUDAF inherited from PythonFuncExpression from which I have moved this code block. I did this so Variant is supported in other UDF types but not UDAFs.
Now, talking specifically about UDAFs, I found an [unofficial blog](https://danvatterott.com/blog/2018/09/06/python-aggregate-udfs-in-pyspark/) to implement PySpark UDFs, so I'll have to look further into it. If it works, I might just remove this code block.
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1831616506
##########
sql/api/src/main/scala/org/apache/spark/sql/util/ArrowUtils.scala:
##########
@@ -146,19 +146,33 @@ private[sql] object ArrowUtils {
ArrowType.Struct.INSTANCE,
null,
Map("variant" -> "true").asJava)
+ val metadataFieldType = new FieldType(
Review Comment:
The line above should not have the variant tag. I have changed the design so that the metadata field has the variant tag. This is because Python codepaths don't have access to the top level field and metadata.
Thanks for catching.
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on code in PR #48770:
URL: https://github.com/apache/spark/pull/48770#discussion_r1831660768
##########
python/pyspark/sql/pandas/types.py:
##########
@@ -1295,6 +1306,16 @@ def convert_udt(value: Any) -> Any:
return convert_udt
+ elif isinstance(dt, VariantType):
+ def convert_variant(variant: Any) -> Any:
Review Comment:
IIUC, This is the inverse of [this function](https://github.com/apache/spark/blob/876d5cab7ffeb55b86516350af23aca8cae35afe/python/pyspark/sql/pandas/types.py#L1043) which converts arrow data to pandas data which can be processed by UDFs. This function converts VariantVal back to arrow so it can be sent downstream via arrow.
Essentially, when variants are received via arrow, the arrow type is struct, so the python type used is dict. This dict is converted to VariantVal using the linked function. The returned VariantVals in UDFs need to be sent downstream using arrow. This function performs this conversion.
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on PR #48770:
URL: https://github.com/apache/spark/pull/48770#issuecomment-2460512184
cc @allisonwang-db @ueshin
--
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
Re: [PR] [SPARK-50238][PYTHON] Add Variant Support in PySpark UDFs/UDTFs [spark]
Posted by "harshmotw-db (via GitHub)" <gi...@apache.org>.
harshmotw-db commented on PR #48770:
URL: https://github.com/apache/spark/pull/48770#issuecomment-2469344317
@gene-db @HyukjinKwon @ueshin I have added support for Variant in Python UDAFs in my [latest commit](https://github.com/apache/spark/pull/48770/commits/8b556bd5ce8c26efa45821debb1113723bbfac02).
--
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