Skip to content

[SPARK-21190][PYSPARK] Python Vectorized UDFs #18659

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
BryanCutler wants to merge 19 commits into from
Closed

[SPARK-21190][PYSPARK] Python Vectorized UDFs #18659

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
be81ef6
vectorized udfs working but hardcoded for ArrowPandasSerializer
BryanCutler Jul 14, 2017
8569736
Added conf for enabling vectorized UDFs, now working
BryanCutler Jul 14, 2017
9236e99
Columns for ArrowPandasSerializer need unique name
BryanCutler Jul 14, 2017
cc7ed5a
fixup ArrowEvalPythonExec to work with new fromPayloadIterator
BryanCutler Aug 4, 2017
cf764b0
changed conf to use a vectorized flag instead on per UDF basis
BryanCutler Sep 6, 2017
4f6c950
fixed chaining of multiple udfs in serializer
BryanCutler Sep 6, 2017
91dead2
changed decorator to pandas_udf
BryanCutler Sep 12, 2017
4a2fec2
changed type to SQL_PANDAS_UDF
BryanCutler Sep 12, 2017
518126e
Add tests.
ueshin Sep 12, 2017
3b4465c
revert unrelated change
BryanCutler Sep 14, 2017
25e3a71
added optional kwarg to pandas_udf to get expected return vector length
BryanCutler Sep 15, 2017
dc237e7
style fixes
BryanCutler Sep 18, 2017
4a0691b
changed kwarg to use length instead of size key
BryanCutler Sep 18, 2017
d49a3db
Added pyarrow.Array type to coerce promoted types, working with pyarr…
BryanCutler Sep 18, 2017
69112a5
Refactor EvalPythonExec.
ueshin Sep 19, 2017
f451d65
fixes from comments in PR
BryanCutler Sep 19, 2017
44a20f6
use generator for conforming series input in serializer
BryanCutler Sep 20, 2017
53926cc
use fillna before casting with astype
BryanCutler Sep 21, 2017
b8ffa50
added test and fix for chained pandas_udf
BryanCutler Sep 21, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
changed type to SQL_PANDAS_UDF
  • Loading branch information
@BryanCutler
BryanCutler committed Sep 12, 2017
commit 4a2fec2aca1998ad04862eeac60d54b088b99de5
2 changes: 1 addition & 1 deletion core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ private[spark] case class ChainedPythonFunctions(funcs: Seq[PythonFunction])
private[spark] object PythonEvalType {
val NON_UDF = 0
val SQL_BATCHED_UDF = 1
val SQL_ARROW_UDF = 2
val SQL_PANDAS_UDF = 2
}

private[spark] object PythonRunner {
Expand Down
2 changes: 1 addition & 1 deletion python/pyspark/serializers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ class SpecialLengths(object):
class PythonEvalType(object):
NON_UDF = 0
SQL_BATCHED_UDF = 1
SQL_ARROW_UDF = 2
SQL_PANDAS_UDF = 2


class Serializer(object):
Expand Down
2 changes: 1 addition & 1 deletion python/pyspark/worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,7 @@ def read_udfs(pickleSer, infile, eval_type):

func = lambda _, it: map(mapper, it)

if eval_type == PythonEvalType.SQL_ARROW_UDF:
if eval_type == PythonEvalType.SQL_PANDAS_UDF:
ser = ArrowPandasSerializer()
else:
ser = BatchedSerializer(PickleSerializer(), 100)
Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/python/ArrowEvalPythonExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,7 +108,7 @@ case class ArrowEvalPythonExec(udfs: Seq[PythonUDF], output: Seq[Attribute], chi

// Output iterator for results from Python.
val outputIterator = new PythonRunner(
pyFuncs, bufferSize, reuseWorker, PythonEvalType.SQL_ARROW_UDF, argOffsets).
pyFuncs, bufferSize, reuseWorker, PythonEvalType.SQL_PANDAS_UDF, argOffsets).
compute(inputIterator, context.partitionId(), context)
Copy link
Member

@viirya viirya Sep 18, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: I think we usually write it in a style as:

val outputIterator = new PythonRunner(
    pyFuncs, bufferSize, reuseWorker, PythonEvalType.SQL_PANDAS_UDF, argOffsets)
  .compute(inputIterator, context.partitionId(), context)

There are similar styles above, e.g. map { case (attr, i) => ... -> .map { case (attr, i) => ....


val joined = new JoinedRow
Expand Down