[SPARK-12717][PYSPARK] Resolving race condition with pyspark broadcasts when using multiple threads #17694

vundela · 2017-04-20T01:46:10Z

What changes were proposed in this pull request?

In pyspark when multiple threads are used, broadcast variables are pickled with wrong PythonRDD wrap functions which leads to the following exception(Because of the race condition between the threads on java side with py4j).

16/01/08 17:10:20 ERROR Executor: Exception in task 0.0 in stage 9.0 (TID 9)
org.apache.spark.api.python.PythonException: Traceback (most recent call last):
File "/Network/Servers/mother.adverplex.com/Volumes/homeland/Users/walker/.spark/spark-1.6.0-bin-hadoop2.6/python/lib/pyspark.zip/pyspark/worker.py", line 98, in main
command = pickleSer._read_with_length(infile)
File "/Network/Servers/mother.adverplex.com/Volumes/homeland/Users/walker/.spark/spark-1.6.0-bin-hadoop2.6/python/lib/pyspark.zip/pyspark/serializers.py", line 164, in _read_with_length
return self.loads(obj)
File "/Network/Servers/mother.adverplex.com/Volumes/homeland/Users/walker/.spark/spark-1.6.0-bin-hadoop2.6/python/lib/pyspark.zip/pyspark/serializers.py", line 422, in loads
return pickle.loads(obj)
File "/Network/Servers/mother.adverplex.com/Volumes/homeland/Users/walker/.spark/spark-1.6.0-bin-hadoop2.6/python/lib/pyspark.zip/pyspark/broadcast.py", line 39, in _from_id
raise Exception("Broadcast variable '%s' not loaded!" % bid)
Exception: (Exception("Broadcast variable '6' not loaded!",), <function _from_id at 0xce7a28>, (6L,))

at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:166)
at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:207)
at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:125)
at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:70)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:270)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
at org.apache.spark.scheduler.Task.run(Task.scala:89)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:213)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)

This change will fix the race condition by making sure that broadcast variables are pickled with same pythonRDD function.

How was this patch tested?

Reproduced the issue mentioned in SPARK-12717, following the instructions specified in jira
Make sure that issue is fixed with the changes.

Please review http://spark.apache.org/contributing.html before opening a pull request.

…ts when using multiple threads

vanzin · 2017-04-20T02:00:30Z

Please remove the boilerplate message at the end of your PR description.

ok to test

maver1ck · 2017-04-20T09:07:22Z

I tested your patch in our environment.

Problem still exists.

Job aborted due to stage failure: Task 0 in stage 22.0 failed 8 times, most recent failure: Lost task 0.7 in stage 22.0 (TID 138, dwh-hn30.adpilot.co): org.apache.spark.api.python.PythonException: Traceback (most recent call last):
  File "/grid/3/hadoop/yarn/log/usercache/bi/appcache/application_1492634694033_0092/container_e538_1492634694033_0092_01_000003/pyspark.zip/pyspark/worker.py", line 161, in main
    func, profiler, deserializer, serializer = read_command(pickleSer, infile)
  File "/grid/3/hadoop/yarn/log/usercache/bi/appcache/application_1492634694033_0092/container_e538_1492634694033_0092_01_000003/pyspark.zip/pyspark/worker.py", line 54, in read_command
    command = serializer._read_with_length(file)
  File "/grid/3/hadoop/yarn/log/usercache/bi/appcache/application_1492634694033_0092/container_e538_1492634694033_0092_01_000003/pyspark.zip/pyspark/serializers.py", line 164, in _read_with_length
    return self.loads(obj)
  File "/grid/3/hadoop/yarn/log/usercache/bi/appcache/application_1492634694033_0092/container_e538_1492634694033_0092_01_000003/pyspark.zip/pyspark/serializers.py", line 419, in loads
    return pickle.loads(obj, encoding=encoding)
  File "/grid/3/hadoop/yarn/log/usercache/bi/appcache/application_1492634694033_0092/container_e538_1492634694033_0092_01_000003/pyspark.zip/pyspark/broadcast.py", line 39, in _from_id
    raise Exception("Broadcast variable '%s' not loaded!" % bid)
Exception: Broadcast variable '22' not loaded!

	at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:193)
	at org.apache.spark.api.python.PythonRunner$$anon$1.<init>(PythonRDD.scala:234)
	at org.apache.spark.api.python.PythonRunner.compute(PythonRDD.scala:152)
	at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:63)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:319)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:283)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:70)
	at org.apache.spark.scheduler.Task.run(Task.scala:86)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
	at java.lang.Thread.run(Thread.java:745)

Driver stacktrace:

maver1ck · 2017-04-20T09:14:46Z

The funny thing is this code works for me on 4 threads and throws exception on 10 threads

vundela · 2017-04-20T18:02:31Z

Hi @maver1ck, Thanks for your time in testing the patch.

I did run the patch with 1000 threads and it works fine. Please check the jira for the log file.

Can you please check /grid/3/hadoop/yarn/log/usercache/bi/appcache/application_1492634694033_0092/container_e538_1492634694033_0092_01_000003/pyspark.zip/pyspark/rdd.py whether the changes are available.

My guess is you might be missing changes in psypark.zip. Can you also let me know the steps you followed to test the patch.

maver1ck · 2017-04-21T08:36:54Z

I checked pyspark.zip of running container and everything is on its place.
So I assume that there is more that one race condition in this code.

I'll try to prepare example of the problem.

maver1ck · 2017-04-21T11:37:00Z

OK. I did additional tests.
Fix is working only with Spark 2.1.
I tried to apply it on 2.0.2 and that was the reason of my problem.

vundela · 2017-04-21T17:58:21Z

Thanks for testing @maver1ck. I will look into 1.6 and 2.0.2.

maver1ck · 2017-04-21T18:03:52Z

@vundela
Great.
But I'm planning to migrate to 2.1 as soon as 2.1.1 will be released.

vundela · 2017-04-21T20:14:34Z

Filed a PR for fixing the issue in spark1.6 branch.

jsoltren · 2017-04-24T18:08:24Z

This seems reasonable to me.

There are some typos in the PR description. I think you meant "pickled" instead of "picked" in a few places.

Using threading.Lock seems okay here from my admittedly limited understanding of the deep details of Python, and my reading of https://docs.python.org/2/library/threading.html#lock-objects.

@vundela and I chatted off thread about this some. The precise race is this: the call to _wrap_function will define a number of broadcast variables. In the time between when the _wrap_function call finishes and self.ctx._jvm.PythonRDD executes, the RDD itself can be modified, perhaps changing broadcast variables and introducing the "Broadcast variable '%s' not loaded!" exception.

My understanding is that, due to the Global Interpreter Lock, this lock will cause all other execution to cease while this block of code runs, implicitly preventing any races. This is a very coarse grained lock for this action but it is as good as we can get. (Someone please correct me if I’m wrong here.)

It would be good if the PR description captured some of the above discussion.

vundela · 2017-04-25T19:42:00Z

cc @holdenk Can you please let me know your comments?

holdenk · 2017-04-25T19:44:57Z

Interesting, I don't think I have all of the required context to review, but I'll try and take a look this weekend (I've got some other things happening this week already).

vundela · 2017-04-25T19:50:24Z

Thanks for your time @holdenk.

vundela · 2017-05-11T20:54:52Z

ping @holdenk, checking if you have some time to review this.

vanzin · 2017-06-07T17:35:37Z

ok to test

SparkQA · 2017-06-07T17:38:56Z

Test build #77800 has finished for PR 17694 at commit 4670678.

This patch fails Python style tests.
This patch merges cleanly.
This patch adds no public classes.

BryanCutler · 2017-07-20T21:56:11Z

Hi @vundela , thanks for your PR. I was able to reproduce your issue, but I think your fix here uses too broad of lock, since the shared resource is the SparkContext pickle registry and needs to be locked only while the python command is being pickled. Please see #18695 and if you wouldn't mind checking if that solves your issue as well. Thanks!

[SPARK-12717][PYSPARK] Resolving race condition with pyspark broadcas…

4670678

…ts when using multiple threads

vundela changed the title ~~[SPARK-12717][PYSPARK] Resolving race condition with pyspark broadcas…~~ [SPARK-12717][PYSPARK] Resolving race condition with pyspark broadcasts when using multiple threads Apr 20, 2017

HyukjinKwon mentioned this pull request Aug 2, 2017

[INFRA] Close stale PRs #18780

Closed

asfgit closed this in 3a45c7f Aug 5, 2017

[SPARK-12717][PYSPARK] Resolving race condition with pyspark broadcasts when using multiple threads #17694

[SPARK-12717][PYSPARK] Resolving race condition with pyspark broadcasts when using multiple threads #17694

Uh oh!

Conversation

vundela commented Apr 20, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

vanzin commented Apr 20, 2017

Uh oh!

maver1ck commented Apr 20, 2017

Uh oh!

maver1ck commented Apr 20, 2017

Uh oh!

vundela commented Apr 20, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

maver1ck commented Apr 21, 2017

Uh oh!

maver1ck commented Apr 21, 2017

Uh oh!

vundela commented Apr 21, 2017

Uh oh!

maver1ck commented Apr 21, 2017

Uh oh!

vundela commented Apr 21, 2017

Uh oh!

jsoltren commented Apr 24, 2017

Uh oh!

vundela commented Apr 25, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

holdenk commented Apr 25, 2017

Uh oh!

vundela commented Apr 25, 2017

Uh oh!

vundela commented May 11, 2017

Uh oh!

vanzin commented Jun 7, 2017

Uh oh!

SparkQA commented Jun 7, 2017

Uh oh!

BryanCutler commented Jul 20, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

7 participants

vundela commented Apr 20, 2017 •

edited

Loading

vundela commented Apr 20, 2017 •

edited

Loading

vundela commented Apr 25, 2017 •

edited

Loading