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

[vundela]

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?

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

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

[vanzin]

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

ok to test

[maver1ck]

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]

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

[vundela]

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]

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]

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]

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

[maver1ck]

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

[vundela]

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

[jsoltren]

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]

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

[holdenk]

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]

Thanks for your time @holdenk.

[vundela]

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

[vanzin]

ok to test

[SparkQA]

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]

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!