Add Support for PythonUDAF expression #937

cindyyuanjiang · 2024-04-13T00:35:45Z

Fixes #936

Changes

Added support for PythonUDAF expression.

Testing
Tested using spark_rapids commands with event log generated locally using pyspark shell. Confirmed behavior is same as PythonUDF support in tools.

Signed-off-by: cindyyuanjiang <[email protected]>

amahussein

Thanks @cindyyuanjiang
I don't believe the PR covers the support for PythonUDAF correctly.

There is a bug in handling PythonUDF. So, it is not a good model to follow. Instead, the PR should fix both of them.
If you have eventlogs that has the PythonUDF/PythonUDAF then this is a good start to look more closely at how the SQLPlanParser handles/detects it.
I had intuition that "PythonUDF/PythonUDAF" won't actually appear in the eventlogs, do they?

amahussein · 2024-04-13T02:54:55Z

core/src/main/scala/org/apache/spark/sql/rapids/tool/ToolUtils.scala

@@ -357,7 +357,7 @@ object RDDCheckHelper {

 object ExecHelper {
  private val UDFRegExLookup = Set(
-    ".*UDF.*".r
+    ".*UDF.*".r, ".*UDAF.*".r


It is correct to add the *UDAF* to account for other UDAFs such as Hive and Scala.
However, the problem that the Q tool uses UDFRegExLookup to decide whether something is uDF or not.
That being said, any node description containing *UDAF* is going to be flagged as not-supported because it is a UDF.
Note that the same problem exists for PythonUDF which is supposed to be supported.

cindyyuanjiang · 2024-04-15T18:32:49Z

There is a bug in handling PythonUDF. So, it is not a good model to follow. Instead, the PR should fix both of them. If you have eventlogs that has the PythonUDF/PythonUDAF then this is a good start to look more closely at how the SQLPlanParser handles/detects it. I had intuition that "PythonUDF/PythonUDAF" won't actually appear in the eventlogs, do they?

Thanks @amahussein! PythonUDF does show up in eventlogs, for example:
"BatchEvalPython [<lambda>(name#1)#6, to_upper(name#1)#8, add_one(age#2L)#9], [pythonUDF0#24, pythonUDF1#25, pythonUDF2#26]"

PythonUDF is not labeled as unsupported in the output, example of rapids_4_spark_qualification_output_unsupportedOperators.csv is as follows:

App ID,SQL ID,Stage ID,ExecId,Unsupported Type,Unsupported Operator,Details,Stage Duration,App Duration,Action
"local-1713205378082",0,3,"0","Exec","BatchEvalPython","Contains UDF",304,138960,"IgnorePerf"
"local-1713205378082",0,3,"1","ReadRDD","Scan ExistingRDD","Is Dataset or RDD",304,138960,"IgnorePerf"
"local-1713205378082",0,0,"2","Exec","BatchEvalPython","Contains UDF",848,138960,"IgnorePerf"
"local-1713205378082",0,1,"3","Exec","BatchEvalPython","Contains UDF",107,138960,"IgnorePerf"
"local-1713205378082",0,2,"4","Exec","BatchEvalPython","Contains UDF",201,138960,"IgnorePerf"
"local-1713205378082",0,-1,"5","Exec","CollectLimit","This is disabled by default because Collect Limit replacement can be slower on the GPU; if huge number of rows in a batch it could help by limiting the number of rows transferred from GPU to CPU",0,138960,"IgnorePerf"
"local-1713205378082",0,-1,"6","Exec","Project","Contains UDF",0,138960,"IgnorePerf"
"local-1713205378082",0,-1,"6","Expr","slen","Is UDF",0,138960,"IgnorePerf"
"local-1713205378082",0,-1,"6","Expr","to_upper","Is UDF",0,138960,"IgnorePerf"
"local-1713205378082",0,-1,"6","Expr","add_one","Is UDF",0,138960,"IgnorePerf"

amahussein · 2024-04-15T18:46:57Z

PythonUDF is not labeled as unsupported in the output

It is not explicitl mentioned but you can see that BatchEvalPython is marked as unsupported because it thought that PythobnUDF is a UDF expr.

"BatchEvalPython","Contains UDF"

That's the part I was mentioning here that the tools was considering PythonUDF as non-supported.

For BatchEvalPython, we should decide if this should be marked as non-/supported. For example, if all PythonUDF expressions appear within BatchEvalPython then we should mark it as supported if it contains PythonUDF.
Q: What did you name your method? If you did not name it, then maybe what we see here is just the name assigned by default from Spark registrar. Are PythonUDF0, PythonUDF1 and PythonUDF2 representing the PythonUDF arguments?

Signed-off-by: cindyyuanjiang <[email protected]>

cindyyuanjiang · 2024-04-16T00:26:01Z

It is not explicitl mentioned but you can see that BatchEvalPython is marked as unsupported because it thought that PythobnUDF is a UDF expr.

Thanks @amahussein!
Made some changes to the PR. To mark pythonUDF as supported:

Distinguish the UDF regex expressions for Scala and Python
For python UDF Execs, mark them as supported and set its unsupported expressions to empty

Now we have python UDF as supported, should we modify skipUDFCheckExecs?
This is an initial approach, opening for discussion. Thanks.

Signed-off-by: cindyyuanjiang <[email protected]>

parthosa · 2024-04-16T17:17:03Z

core/src/main/scala/com/nvidia/spark/rapids/tool/planparser/SQLPlanParser.scala

@@ -313,7 +315,7 @@ object ExecInfo {
      duration,
      nodeId,
      opType,
-      isSupported,
+      finalSupported,
      children,
      stages,
      shouldRemove,


Do we need to make changes to the usage of variable containsUDF when passing to createExecNoNode()

I don't think so. Now containsUDF denotes that the exec contains Scala UDFs after we changed the matching regex, so for python UDFs, this value would remain false as intended.

Signed-off-by: cindyyuanjiang <[email protected]>

parthosa

Thank you @cindyyuanjiang for this change.

amahussein

Thanks @cindyyuanjiang

I cannot see how the changes will treat pythonUDAF as supported. All the changes are related to pythonUDF
I don't see how the Q tool shows the batchEvalPython as supported. That means that the changes done in ExecHelper did change the behavior of the flow. Note that pythonUDF/PythonUDAF are expressions and we should not expect to see the Q tool changes the decision about an exec (batchEvalPython) without main changes to the Exec parser. If you confirm that batchEvalPython is always linked to pythonUDF/PythonUDAF, then we should simply add code to catch that exec and mark it as supported without need to parse the content.

CC: @nartal1 any thoughts?

amahussein · 2024-04-16T19:15:02Z

core/src/main/scala/org/apache/spark/sql/rapids/tool/AppBase.scala

@@ -331,7 +331,7 @@ abstract class AppBase(
    }
  }

-  private val UDFRegex = ".*UDF.*"
+  private val UDFRegex = ".*(?<!python)UDF.*"


this won't match pythonUDAF

amahussein · 2024-04-16T19:19:09Z

core/src/main/scala/org/apache/spark/sql/rapids/tool/ToolUtils.scala

+    ".*(?<!python)UDF.*".r
+  )
+
+  private val pythonUDFRegExLookUp = Set(


Same here as my previous comment. There are three regular expressions that look overlapping.

amahussein · 2024-04-16T19:19:52Z

core/src/main/scala/org/apache/spark/sql/rapids/tool/ToolUtils.scala

+  )
+
+  private val pythonUDFRegExLookUp = Set(
+    ".*pythonUDF.*".r


This does not take pythonUDAF into considerations.

amahussein · 2024-04-16T19:24:02Z

core/src/test/scala/com/nvidia/spark/rapids/tool/planparser/SqlPlanParserSuite.scala

+    val projectExec = execInfo.filter(_.exec.contains("Project"))
+    assertSizeAndSupported(2, projectExec)
+    val batchEvalPythonExec = execInfo.filter(_.exec.contains("BatchEvalPython"))
+    assertSizeAndSupported(1, batchEvalPythonExec)


there is something I missing here. I cannot find anywhere in the code that we handle BatchEvalPython exec.
How come that the result shows it as supported?

BatchEvalPython should not be supported. I will fix this.

Have we confirmed that for a fact?
I mean did we verify that it is most likely that BatchEvalPython is associated with PythonUDF and PythonUDAF? if that's the case then all we need to do is considering that the entire exec is supported with being conservative on the speedup factor.
As far as I can see in the plugin code, the RAPIDS has very limited capability accelerating those expressions anyway. So, the speedup should be near 1.

amahussein · 2024-04-16T19:30:22Z

core/src/main/scala/com/nvidia/spark/rapids/tool/planparser/SQLPlanParser.scala


    // if the expression is RDD because of the node name, then we do not want to add the
    // unsupportedExpressions because it becomes bogus.
-    val finalUnsupportedExpr = if (rddCheckRes.nodeDescRDD) {
+    val finalUnsupportedExpr = if (rddCheckRes.nodeDescRDD || containsPythonUDF) {


I don't think this is correct.
The code comment just above that change is explaining that we avoid parsing the node content when it is rddCheckRes.nodeDescRDD . But the new code tries to check if it is pythonUDF anyway which could lead to bogus results.

Sorry for the confusion. I will address this in the next iteration.

nartal1 · 2024-04-16T20:43:29Z

I don't see how the Q tool shows the batchEvalPython as supported. That means that the changes done in ExecHelper did change the behavior of the flow. Note that pythonUDF/PythonUDAF are expressions and we should not expect to see the Q tool changes the decision about an exec (batchEvalPython) without main changes to the Exec parser. If you confirm that batchEvalPython is always linked to pythonUDF/PythonUDAF, then we should simply add code to catch that exec and mark it as supported without need to parse the content.

I am confused how batchEvalPython is marked as supported and is passing the test.
@cindyyuanjiang - Could you please confirm if batchEvalPython is linked with either pythonUDF/PythonUDAF as per @amahussein suggestion? I think adding a new parser for batchEvalPython makes sense in that case.

amahussein

Let's be more conservative with the speedup value we assign for PythonUDF and PythonUDAF.
I would suggest a speedup-factor closer to 1.0 actually.

Signed-off-by: cindyyuanjiang <[email protected]>

cindyyuanjiang · 2024-04-17T18:51:27Z

Thanks @amahussein @nartal1 for the feedback!
I made some updates to the logic for handling pythonUDFs and updated the speedup factors.

pythonUDAF
I made multiple attempts to generate event logs for pythonUDAFs but this operator name never show up. The ones I generated included AggregateInPandas which is already handled by existing implementation, and will be marked as supported. We can have further discussion on how to improve this.

pythonUDF
I separated the regex matching patterns for Scala and python so that we can use function isUDF for Scala UDF and isPythonUDF for python UDF.
For Exec that contains pythonUDF in its description, this PR marks it as supported if the Exec is supported in the CSV even if it might have unsupported expressions which can be UDFs. This is an estimation and optimistic. This approach requires us to add this logic for individual exec parsers, this PR currently only handles ProjectExec, open this for discussion.

Note:
BatchEvalPython is a physical plan node specifically designed to handle the batch execution of Python UDFs. This is not supported by the plugin (! <BatchEvalPythonExec> cannot run on GPU because GPU does not currently support the operator class org.apache.spark.sql.execution.python.BatchEvalPythonExec).

Followup: had an offline discussion. I will add a parser for BatchEvalPython.

parthosa · 2024-04-17T20:55:08Z

Yes, It seems that all PythonUDAF expressions gets converted to AggregateInPandasExec

https://github.com/apache/spark/blob/4957a40d6e6bf68226c8047687e8f30c93adb8ce/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala#L640-L646

Signed-off-by: cindyyuanjiang <[email protected]>

cindyyuanjiang added 2 commits April 12, 2024 17:30

added pythonUDAF support

6d651b6

Signed-off-by: cindyyuanjiang <[email protected]>

removed pythonudaf from override configs file

cadb525

Signed-off-by: cindyyuanjiang <[email protected]>

cindyyuanjiang requested review from amahussein, parthosa and nartal1 April 13, 2024 00:38

cindyyuanjiang self-assigned this Apr 13, 2024

cindyyuanjiang added bug Something isn't working core_tools Scope the core module (scala) labels Apr 13, 2024

amahussein requested changes Apr 13, 2024

View reviewed changes

added pythonudf support

f9701a1

Signed-off-by: cindyyuanjiang <[email protected]>

added test case for python udf

d70668a

Signed-off-by: cindyyuanjiang <[email protected]>

cindyyuanjiang requested a review from amahussein April 16, 2024 01:21

parthosa reviewed Apr 16, 2024

View reviewed changes

removed testing logging

297c7ad

Signed-off-by: cindyyuanjiang <[email protected]>

cindyyuanjiang requested a review from parthosa April 16, 2024 19:02

parthosa previously approved these changes Apr 16, 2024

View reviewed changes

amahussein requested changes Apr 16, 2024

View reviewed changes

amahussein reviewed Apr 17, 2024

View reviewed changes

cindyyuanjiang added 2 commits April 17, 2024 10:53

updated logic

8a9fec9

Signed-off-by: cindyyuanjiang <[email protected]>

Merge branch 'dev' into spark-rapids-tools-936

e542bf3

cindyyuanjiang dismissed parthosa’s stale review via e542bf3 April 17, 2024 18:10

updated speedup factor for python udf and udafs

cfae1a8

Signed-off-by: cindyyuanjiang <[email protected]>

added a simple parser for batchevalpython exec

948a344

Signed-off-by: cindyyuanjiang <[email protected]>

cindyyuanjiang marked this pull request as draft April 17, 2024 22:10

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add Support for PythonUDAF expression #937

Add Support for PythonUDAF expression #937

cindyyuanjiang commented Apr 13, 2024 •

edited

Loading

amahussein left a comment

amahussein Apr 13, 2024

cindyyuanjiang commented Apr 15, 2024 •

edited

Loading

amahussein commented Apr 15, 2024

cindyyuanjiang commented Apr 16, 2024

parthosa Apr 16, 2024 •

edited

Loading

cindyyuanjiang Apr 16, 2024

parthosa left a comment

amahussein left a comment

amahussein Apr 16, 2024

amahussein Apr 16, 2024

amahussein Apr 16, 2024

amahussein Apr 16, 2024

cindyyuanjiang Apr 16, 2024

amahussein Apr 17, 2024

amahussein Apr 16, 2024

cindyyuanjiang Apr 16, 2024

nartal1 commented Apr 16, 2024

amahussein left a comment

cindyyuanjiang commented Apr 17, 2024 •

edited

Loading

parthosa commented Apr 17, 2024 •

edited

Loading

Add Support for PythonUDAF expression #937

Are you sure you want to change the base?

Add Support for PythonUDAF expression #937

Conversation

cindyyuanjiang commented Apr 13, 2024 • edited Loading

amahussein left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cindyyuanjiang commented Apr 15, 2024 • edited Loading

amahussein commented Apr 15, 2024

cindyyuanjiang commented Apr 16, 2024

parthosa Apr 16, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

parthosa left a comment

Choose a reason for hiding this comment

amahussein left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nartal1 commented Apr 16, 2024

amahussein left a comment

Choose a reason for hiding this comment

cindyyuanjiang commented Apr 17, 2024 • edited Loading

parthosa commented Apr 17, 2024 • edited Loading

cindyyuanjiang commented Apr 13, 2024 •

edited

Loading

cindyyuanjiang commented Apr 15, 2024 •

edited

Loading

parthosa Apr 16, 2024 •

edited

Loading

cindyyuanjiang commented Apr 17, 2024 •

edited

Loading

parthosa commented Apr 17, 2024 •

edited

Loading