[SPARK-23177][SQL][PySpark] Extract zero-parameter UDFs from aggregate #20360

viirya · 2018-01-23T07:14:26Z

What changes were proposed in this pull request?

We extract Python UDFs in logical aggregate which depends on aggregate expression or grouping key in ExtractPythonUDFFromAggregate rule. But Python UDFs which don't depend on above expressions should also be extracted to avoid the issue reported in the JIRA.

A small code snippet to reproduce that issue looks like:

import pyspark.sql.functions as f

df = spark.createDataFrame([(1,2), (3,4)])
f_udf = f.udf(lambda: str("const_str"))
df2 = df.distinct().withColumn("a", f_udf())
df2.show()

Error exception is raised as:

: org.apache.spark.sql.catalyst.errors.package$TreeNodeException: Binding attribute, tree: pythonUDF0#50
        at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:56)
        at org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:91)
        at org.apache.spark.sql.catalyst.expressions.BindReferences$$anonfun$bindReference$1.applyOrElse(BoundAttribute.scala:90)
        at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)
        at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$2.apply(TreeNode.scala:267)
        at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:70)
        at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:266)
        at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformDown$1.apply(TreeNode.scala:272)
        at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$transformDown$1.apply(TreeNode.scala:272)
        at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:306)
        at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)
        at org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:304)
        at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:272)
        at org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:256)
        at org.apache.spark.sql.catalyst.expressions.BindReferences$.bindReference(BoundAttribute.scala:90)
        at org.apache.spark.sql.execution.aggregate.HashAggregateExec$$anonfun$38.apply(HashAggregateExec.scala:514)
        at org.apache.spark.sql.execution.aggregate.HashAggregateExec$$anonfun$38.apply(HashAggregateExec.scala:513)

This exception raises because HashAggregateExec tries to bind the aliased Python UDF expression (e.g., pythonUDF0#50 AS a#44) to grouping key.

How was this patch tested?

Added test.

SparkQA · 2018-01-23T07:19:44Z

Test build #86518 has finished for PR 20360 at commit b6cb621.

This patch fails Python style tests.
This patch does not merge cleanly.
This patch adds no public classes.

viirya · 2018-01-23T07:28:45Z

cc @HyukjinKwon @cloud-fan

SparkQA · 2018-01-23T08:05:01Z

Test build #86520 has finished for PR 20360 at commit 5c3afbb.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2018-01-23T08:36:44Z

retest this please.

ueshin

LGTM. I left one question though.

ueshin · 2018-01-23T09:45:25Z

sql/core/src/main/scala/org/apache/spark/sql/execution/python/ExtractPythonUDFs.scala

@@ -45,7 +45,8 @@ object ExtractPythonUDFFromAggregate extends Rule[LogicalPlan] {

  private def hasPythonUdfOverAggregate(expr: Expression, agg: Aggregate): Boolean = {
    expr.find {
-      e => PythonUDF.isScalarPythonUDF(e) && e.find(belongAggregate(_, agg)).isDefined
+      e => PythonUDF.isScalarPythonUDF(e) &&
+        (e.references.isEmpty || e.find(belongAggregate(_, agg)).isDefined)


Can we use just e.children instead of e.references?

I just want to consider some literal inputs like df2 = df.distinct().withColumn("a", f_udf(f.lit("2"))).

Sorry, I wrote a duplicate comment and removed it back. It didn't show up when I write ..

Oh, I see, sounds good. Thanks!

SparkQA · 2018-01-23T11:56:08Z

Test build #86523 has finished for PR 20360 at commit 5c3afbb.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HyukjinKwon · 2018-01-23T12:34:28Z

LGTM

cloud-fan · 2018-01-23T13:02:29Z

sql/core/src/main/scala/org/apache/spark/sql/execution/python/ExtractPythonUDFs.scala

@@ -45,7 +45,8 @@ object ExtractPythonUDFFromAggregate extends Rule[LogicalPlan] {

  private def hasPythonUdfOverAggregate(expr: Expression, agg: Aggregate): Boolean = {
    expr.find {
-      e => PythonUDF.isScalarPythonUDF(e) && e.find(belongAggregate(_, agg)).isDefined


shall we update the classdoc too? it currently says Extracts all the Python UDFs in logical aggregate, which depends on aggregate expression or grouping key, evaluate them after aggregate

Yes. Updated.

cloud-fan · 2018-01-23T13:06:18Z

LGTM

SparkQA · 2018-01-24T02:41:46Z

Test build #86551 has finished for PR 20360 at commit 74684a7.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

HyukjinKwon · 2018-01-24T02:44:58Z

Merged to master.

HyukjinKwon · 2018-01-24T02:45:11Z

@viirya, mind if I ask to open a backport to branch-2.3?

viirya · 2018-01-24T03:13:59Z

@HyukjinKwon Ok. I will open a backport later.

hankim · 2018-04-09T18:24:05Z

is there any workaround for this? my environment hasn't upgrade to 2.3.0, but I have exact code that jira ticket has. (http://mail-archives.apache.org/mod_mbox/spark-issues/201801.mbox/%3CJIRA.13132665.1516622460000.6681.1516622520346@Atlassian.JIRA%3E)
i.e., assigning uuid after distinct() call with udf.
Thank you!
cc @viirya @HyukjinKwon @cloud-fan

viirya · 2018-04-10T07:38:25Z

@hankim maybe like:

import pyspark.sql.functions as f
import uuid

df = spark.createDataFrame([(1,2), (3,4)])
f_udf = f.udf(lambda: str(uuid.uuid4()))
df2 = df.distinct().cache()
df3 = df2.withColumn("a", f_udf()).show()

Extract parameter-less UDFs from aggregate.

b6cb621

viirya added 2 commits January 23, 2018 07:22

Merge remote-tracking branch 'upstream/master' into SPARK-23177

f1db144

Fix python style.

5c3afbb

ueshin reviewed Jan 23, 2018

View reviewed changes

cloud-fan reviewed Jan 23, 2018

View reviewed changes

Fix doc.

74684a7

asfgit closed this in a3911cf Jan 24, 2018

viirya deleted the SPARK-23177 branch December 27, 2023 18:21

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-23177][SQL][PySpark] Extract zero-parameter UDFs from aggregate #20360

[SPARK-23177][SQL][PySpark] Extract zero-parameter UDFs from aggregate #20360

viirya commented Jan 23, 2018

SparkQA commented Jan 23, 2018

viirya commented Jan 23, 2018

SparkQA commented Jan 23, 2018

viirya commented Jan 23, 2018

ueshin left a comment

ueshin Jan 23, 2018

viirya Jan 23, 2018

HyukjinKwon Jan 23, 2018

ueshin Jan 23, 2018

SparkQA commented Jan 23, 2018

HyukjinKwon commented Jan 23, 2018

cloud-fan Jan 23, 2018

viirya Jan 23, 2018

cloud-fan commented Jan 23, 2018

SparkQA commented Jan 24, 2018

HyukjinKwon commented Jan 24, 2018

HyukjinKwon commented Jan 24, 2018

viirya commented Jan 24, 2018

hankim commented Apr 9, 2018

viirya commented Apr 10, 2018

[SPARK-23177][SQL][PySpark] Extract zero-parameter UDFs from aggregate #20360

[SPARK-23177][SQL][PySpark] Extract zero-parameter UDFs from aggregate #20360

Conversation

viirya commented Jan 23, 2018

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Jan 23, 2018

viirya commented Jan 23, 2018

SparkQA commented Jan 23, 2018

viirya commented Jan 23, 2018

ueshin left a comment

Choose a reason for hiding this comment

ueshin Jan 23, 2018

Choose a reason for hiding this comment

viirya Jan 23, 2018

Choose a reason for hiding this comment

HyukjinKwon Jan 23, 2018

Choose a reason for hiding this comment

ueshin Jan 23, 2018

Choose a reason for hiding this comment

SparkQA commented Jan 23, 2018

HyukjinKwon commented Jan 23, 2018

cloud-fan Jan 23, 2018

Choose a reason for hiding this comment

viirya Jan 23, 2018

Choose a reason for hiding this comment

cloud-fan commented Jan 23, 2018

SparkQA commented Jan 24, 2018

HyukjinKwon commented Jan 24, 2018

HyukjinKwon commented Jan 24, 2018

viirya commented Jan 24, 2018

hankim commented Apr 9, 2018

viirya commented Apr 10, 2018