[SPARK-33277][PYSPARK][SQL] Use ContextAwareIterator to stop consuming after the task ends. #30242

ueshin · 2020-11-04T00:54:14Z

What changes were proposed in this pull request?

This is a retry of #30177.

Makes TaskCompletion event thread wait until the thread ends to avoid the race condition.

Why are the changes needed?

There are still sometimes crashes of executors as discussed at #30177 (comment).

The race condition could happen between !context.isCompleted() && !context.isInterrupted() and iter.hasNext in the hasNext method.
This is because the TaskCompletion event thread could close the upstream iterator even between them.
We should make the event wait for a while until the consuming thread ends which should end soon as the iterator returns false in hasNext.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing tests.

SparkQA · 2020-11-04T01:44:28Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35183/

SparkQA · 2020-11-04T02:07:05Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35183/

SparkQA · 2020-11-04T02:51:13Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35185/

SparkQA · 2020-11-04T03:20:26Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35185/

SparkQA · 2020-11-04T03:51:19Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35186/

SparkQA · 2020-11-04T04:12:24Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35186/

SparkQA · 2020-11-04T04:26:27Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35189/

SparkQA · 2020-11-04T04:51:58Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35189/

SparkQA · 2020-11-04T05:11:07Z

Test build #130582 has finished for PR 30242 at commit 3f96145.

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

SparkQA · 2020-11-04T06:06:27Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35191/

SparkQA · 2020-11-04T06:27:58Z

Test build #130584 has finished for PR 30242 at commit 2a0d2af.

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

SparkQA · 2020-11-04T06:32:43Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35191/

SparkQA · 2020-11-04T07:29:51Z

Test build #130586 has finished for PR 30242 at commit 6e5be90.

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

SparkQA · 2020-11-04T08:05:01Z

Test build #130589 has finished for PR 30242 at commit 895d91d.

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

SparkQA · 2020-11-04T08:05:01Z

Test build #130591 has finished for PR 30242 at commit 997e1aa.

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

SparkQA · 2020-11-04T20:36:59Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35217/

SparkQA · 2020-11-04T20:54:57Z

Test build #130616 has finished for PR 30242 at commit aec13c2.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-11-04T20:58:42Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35217/

ueshin · 2020-11-04T21:05:56Z

Jenkins, retest this please.

SparkQA · 2020-11-04T22:03:45Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35220/

SparkQA · 2020-11-04T22:25:06Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35220/

SparkQA · 2020-11-06T01:20:35Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35288/

SparkQA · 2020-11-06T01:52:20Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/35288/

SparkQA · 2020-11-06T02:04:14Z

Test build #130668 has finished for PR 30242 at commit 8b11647.

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

SparkQA · 2020-11-06T03:18:58Z

Test build #130674 has finished for PR 30242 at commit 1ba0b65.

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

ueshin · 2020-11-06T03:22:27Z

Now PySpark tests seem not fail.
@HyukjinKwon @dongjoon-hyun Could you take a look at this? Thanks!

SparkQA · 2020-11-06T05:11:24Z

Test build #130677 has finished for PR 30242 at commit 87d8854.

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

HyukjinKwon · 2020-11-06T06:49:12Z

cc @zsxwing too

ueshin · 2020-11-14T02:39:15Z

gentle ping

gatorsmile · 2020-11-22T05:10:14Z

ping @zsxwing

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

zsxwing · 2020-11-23T17:50:51Z

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

+
+      while (thread == null && !failed.get()) {
+        // Wait for a while since the writer thread might not reach to consuming the iterator yet.
+        context.wait(10)


Did you mean Thread.sleep(10)? Object.wait is not supposed to use like this.

I do mean wait. This will run within synchronized(context) and we should release the lock for the writer thread while waiting.

I didn't realize it. It's better to not rely on this in a listener. This is something we should consider to improve in future. It's a bad idea to hold an implicit lock when calling user's listener because it's pretty easy to cause surprising deadlock.

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

zsxwing · 2020-11-23T18:05:02Z

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

+
+  val thread = new AtomicReference[Thread]()
+
+  if (iter.hasNext) {


Will this change the thread that iter.hasNext is running? We can add the listeners without checking it.

Actually this is to make sure the upstream iterator is initialized. The upstream iterator must be initialized earlier as it might register another completion listener and the listener should run later than this one.

zsxwing · 2020-11-23T18:35:37Z

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

+      failed.set(true)
+    }
+
+    context.addTaskCompletionListener[Unit] { _ =>


This assumes the task completion listener to stop thread runs before this one. Otherwise, it would hang forever. I'm wondering if there is any better solution to avoid this implicit assumption.

The task completion lister will wait for the thread to stop within this listener, and the thread will stop soon as it checks !context.isCompleted() && !context.isInterrupted().

SparkQA · 2020-11-25T07:02:52Z

Test build #131717 has finished for PR 30242 at commit e2cc227.

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

cloud-fan · 2020-12-03T10:03:04Z

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

+        // Use `context.wait()` instead of `Thread.sleep()` here since the task completion lister
+        // works under `synchronized(context)`. We might need to consider to improve in the future.
+        // It's a bad idea to hold an implicit lock when calling user's listener because it's
+        // pretty easy to cause surprising deadlock.


This is a bit scary. Is there a better way?

It's a bad idea to hold an implicit lock when calling user's listener because it's pretty easy to cause surprising deadlock.

Maybe we can fix this first. The this listener doesn't need to rely on an implicit lock.

I see. Let me change the strategy here.

…g after the task ends ### What changes were proposed in this pull request? This is a retry of #30177. This is not a complete fix, but it would take long time to complete (#30242). As discussed offline, at least using `ContextAwareIterator` should be helpful enough for many cases. As the Python evaluation consumes the parent iterator in a separate thread, it could consume more data from the parent even after the task ends and the parent is closed. Thus, we should use `ContextAwareIterator` to stop consuming after the task ends. ### Why are the changes needed? Python/Pandas UDF right after off-heap vectorized reader could cause executor crash. E.g.,: ```py spark.range(0, 100000, 1, 1).write.parquet(path) spark.conf.set("spark.sql.columnVector.offheap.enabled", True) def f(x): return 0 fUdf = udf(f, LongType()) spark.read.parquet(path).select(fUdf('id')).head() ``` This is because, the Python evaluation consumes the parent iterator in a separate thread and it consumes more data from the parent even after the task ends and the parent is closed. If an off-heap column vector exists in the parent iterator, it could cause segmentation fault which crashes the executor. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added tests, and manually. Closes #30899 from ueshin/issues/SPARK-33277/context_aware_iterator. Authored-by: Takuya UESHIN <[email protected]> Signed-off-by: Dongjoon Hyun <[email protected]>

…g after the task ends ### What changes were proposed in this pull request? This is a retry of #30177. This is not a complete fix, but it would take long time to complete (#30242). As discussed offline, at least using `ContextAwareIterator` should be helpful enough for many cases. As the Python evaluation consumes the parent iterator in a separate thread, it could consume more data from the parent even after the task ends and the parent is closed. Thus, we should use `ContextAwareIterator` to stop consuming after the task ends. ### Why are the changes needed? Python/Pandas UDF right after off-heap vectorized reader could cause executor crash. E.g.,: ```py spark.range(0, 100000, 1, 1).write.parquet(path) spark.conf.set("spark.sql.columnVector.offheap.enabled", True) def f(x): return 0 fUdf = udf(f, LongType()) spark.read.parquet(path).select(fUdf('id')).head() ``` This is because, the Python evaluation consumes the parent iterator in a separate thread and it consumes more data from the parent even after the task ends and the parent is closed. If an off-heap column vector exists in the parent iterator, it could cause segmentation fault which crashes the executor. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added tests, and manually. Closes #30899 from ueshin/issues/SPARK-33277/context_aware_iterator. Authored-by: Takuya UESHIN <[email protected]> Signed-off-by: Dongjoon Hyun <[email protected]> (cherry picked from commit 5c9b421) Signed-off-by: Dongjoon Hyun <[email protected]>

…g after the task ends ### What changes were proposed in this pull request? This is a retry of apache#30177. This is not a complete fix, but it would take long time to complete (apache#30242). As discussed offline, at least using `ContextAwareIterator` should be helpful enough for many cases. As the Python evaluation consumes the parent iterator in a separate thread, it could consume more data from the parent even after the task ends and the parent is closed. Thus, we should use `ContextAwareIterator` to stop consuming after the task ends. ### Why are the changes needed? Python/Pandas UDF right after off-heap vectorized reader could cause executor crash. E.g.,: ```py spark.range(0, 100000, 1, 1).write.parquet(path) spark.conf.set("spark.sql.columnVector.offheap.enabled", True) def f(x): return 0 fUdf = udf(f, LongType()) spark.read.parquet(path).select(fUdf('id')).head() ``` This is because, the Python evaluation consumes the parent iterator in a separate thread and it consumes more data from the parent even after the task ends and the parent is closed. If an off-heap column vector exists in the parent iterator, it could cause segmentation fault which crashes the executor. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added tests, and manually.

…suming after the task ends ### What changes were proposed in this pull request? This is a backport of #30899. This is not a complete fix, but it would take long time to complete (#30242). As discussed offline, at least using `ContextAwareIterator` should be helpful enough for many cases. As the Python evaluation consumes the parent iterator in a separate thread, it could consume more data from the parent even after the task ends and the parent is closed. Thus, we should use `ContextAwareIterator` to stop consuming after the task ends. ### Why are the changes needed? Python/Pandas UDF right after off-heap vectorized reader could cause executor crash. E.g.,: ```py spark.range(0, 100000, 1, 1).write.parquet(path) spark.conf.set("spark.sql.columnVector.offheap.enabled", True) def f(x): return 0 fUdf = udf(f, LongType()) spark.read.parquet(path).select(fUdf('id')).head() ``` This is because, the Python evaluation consumes the parent iterator in a separate thread and it consumes more data from the parent even after the task ends and the parent is closed. If an off-heap column vector exists in the parent iterator, it could cause segmentation fault which crashes the executor. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added tests, and manually. Closes #30913 from ueshin/issues/SPARK-33277/2.4/context_aware_iterator. Authored-by: Takuya UESHIN <[email protected]> Signed-off-by: HyukjinKwon <[email protected]>

github-actions · 2021-03-14T00:50:24Z

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!

Block TaskCompletion event until the thread ends.

3f96145

ueshin requested a review from HyukjinKwon November 4, 2020 00:54

Fix.

2a0d2af

Fix.

6e5be90

Comments.

895d91d

ueshin force-pushed the issues/SPARK-33277/block branch from 676c530 to 895d91d Compare November 4, 2020 03:40

Fix.

997e1aa

Fix.

aec13c2

Run tests.

7db9bb8

Run tests.

87d8854

ueshin marked this pull request as ready for review November 6, 2020 03:22

zsxwing requested changes Nov 23, 2020

View reviewed changes

ueshin added 2 commits November 24, 2020 17:19

Merge branch 'master' into issues/SPARK-33277/block

c77b61e

Add private.

e2cc227

ueshin added 3 commits December 1, 2020 14:52

Merge branch 'master' into issues/SPARK-33277/block

ee308de

Add more comments.

429c159

Rerun tests.

46613dd

cloud-fan reviewed Dec 3, 2020

View reviewed changes

ueshin marked this pull request as draft December 3, 2020 20:38

ueshin mentioned this pull request Dec 23, 2020

[SPARK-33277][PYSPARK][SQL] Use ContextAwareIterator to stop consuming after the task ends. #30899

Closed

ueshin mentioned this pull request Dec 23, 2020

[SPARK-33277][PYSPARK][SQL][2.4] Use ContextAwareIterator to stop consuming after the task ends. #30913

Closed

github-actions bot added the Stale label Mar 14, 2021

github-actions bot closed this Mar 15, 2021


		val thread = new AtomicReference[Thread]()

		if (iter.hasNext) {

[SPARK-33277][PYSPARK][SQL] Use ContextAwareIterator to stop consuming after the task ends. #30242

[SPARK-33277][PYSPARK][SQL] Use ContextAwareIterator to stop consuming after the task ends. #30242

Uh oh!

Conversation

ueshin commented Nov 4, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

ueshin commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 4, 2020

Uh oh!

SparkQA commented Nov 6, 2020

Uh oh!

SparkQA commented Nov 6, 2020

Uh oh!

SparkQA commented Nov 6, 2020

Uh oh!

SparkQA commented Nov 6, 2020

Uh oh!

ueshin commented Nov 6, 2020

Uh oh!

SparkQA commented Nov 6, 2020

Uh oh!

HyukjinKwon commented Nov 6, 2020

Uh oh!

ueshin commented Nov 14, 2020

Uh oh!

gatorsmile commented Nov 22, 2020

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

ueshin commented Nov 4, 2020 •

edited

Loading