[SPARK-23754][PYTHON][FOLLOWUP][BACKPORT-2.3] Move UDF stop iteration wrapping from driver to executor #21538
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Make sure that `StopIteration`s raised in users' code do not silently interrupt processing by spark, but are raised as exceptions to the users. The users' functions are wrapped in `safe_iter` (in `shuffle.py`), which re-raises `StopIteration`s as `RuntimeError`s Unit tests, making sure that the exceptions are indeed raised. I am not sure how to check whether a `Py4JJavaError` contains my exception, so I simply looked for the exception message in the java exception's `toString`. Can you propose a better way? This is my original work, licensed in the same way as spark Author: e-dorigatti <emilio.dorigatti@gmail.com> Closes apache#21383 from e-dorigatti/fix_spark_23754. (cherry picked from commit 0ebb0c0)
… driver to executor SPARK-23754 was fixed in apache#21383 by changing the UDF code to wrap the user function, but this required a hack to save its argspec. This PR reverts this change and fixes the `StopIteration` bug in the worker The root of the problem is that when an user-supplied function raises a `StopIteration`, pyspark might stop processing data, if this function is used in a for-loop. The solution is to catch `StopIteration`s exceptions and re-raise them as `RuntimeError`s, so that the execution fails and the error is reported to the user. This is done using the `fail_on_stopiteration` wrapper, in different ways depending on where the function is used: - In RDDs, the user function is wrapped in the driver, because this function is also called in the driver itself. - In SQL UDFs, the function is wrapped in the worker, since all processing happens there. Moreover, the worker needs the signature of the user function, which is lost when wrapping it, but passing this signature to the worker requires a not so nice hack. Same tests, plus tests for pandas UDFs Author: edorigatti <emilio.dorigatti@gmail.com> Closes apache#21467 from e-dorigatti/fix_udf_hack.
Member
|
@e-dorigatti Can you add |
e-dorigatti
changed the title
Member
|
add to whitelist |
|
Test build #91701 has finished for PR 21538 at commit
|
Contributor
Author
|
Seems like it skipped the pandas tests, for both python2.7 and pypy |
Member
|
Yea, it's unfortunate .. we should fix and set up the Jenkins env too. |
|
Test build #91704 has finished for PR 21538 at commit
|
HyukjinKwon
reviewed
Jun 12, 2018
python/pyspark/worker.py
Outdated
|
|
||
| # make sure StopIteration's raised in the user code are not ignored | ||
| # when they are processed in a for loop, raise them as RuntimeError's instead | ||
| row_func = fail_on_stopiteration(row_func) |
Member
There was a problem hiding this comment.
@e-dorigatti, I think it's fine to name it func as fixed in master. Let's reduce the diff so that other backports make less conflicts in the future.
|
Test build #91716 has finished for PR 21538 at commit
|
Member
|
Merged to branch-2.3. |
asfgit
pushed a commit
that referenced
this pull request
Jun 13, 2018
… wrapping from driver to executor SPARK-23754 was fixed in #21383 by changing the UDF code to wrap the user function, but this required a hack to save its argspec. This PR reverts this change and fixes the `StopIteration` bug in the worker. The root of the problem is that when an user-supplied function raises a `StopIteration`, pyspark might stop processing data, if this function is used in a for-loop. The solution is to catch `StopIteration`s exceptions and re-raise them as `RuntimeError`s, so that the execution fails and the error is reported to the user. This is done using the `fail_on_stopiteration` wrapper, in different ways depending on where the function is used: - In RDDs, the user function is wrapped in the driver, because this function is also called in the driver itself. - In SQL UDFs, the function is wrapped in the worker, since all processing happens there. Moreover, the worker needs the signature of the user function, which is lost when wrapping it, but passing this signature to the worker requires a not so nice hack. HyukjinKwon Author: edorigatti <emilio.dorigatti@gmail.com> Author: e-dorigatti <emilio.dorigatti@gmail.com> Closes #21538 from e-dorigatti/branch-2.3.
Member
|
@e-dorigatti, this got merged into branch-2.3. Likewise, this also should be manually closed. Thanks for working on this. |
Contributor
Author
|
@HyukjinKwon thank you so much for your patience :) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
SPARK-23754 was fixed in #21383 by changing the UDF code to wrap the user function, but this required a hack to save its argspec. This PR reverts this change and fixes the
StopIterationbug in the worker.The root of the problem is that when an user-supplied function raises a
StopIteration, pyspark might stop processing data, if this function is used in a for-loop. The solution is to catchStopIterations exceptions and re-raise them asRuntimeErrors, so that the execution fails and the error is reported to the user. This is done using thefail_on_stopiterationwrapper, in different ways depending on where the function is used:@HyukjinKwon