title: Tips for properly using large broadcast variables?
body: <p>I'm using a broadcast variable about 100 MB pickled in size, which I'm approximating with:</p>
  
  <pre><code>&gt;&gt;&gt; data = list(range(int(10*1e6)))
  &gt;&gt;&gt; import cPickle as pickle
  &gt;&gt;&gt; len(pickle.dumps(data))
  98888896
  </code></pre>
  
  <p>Running on a cluster with 3 c3.2xlarge executors, and a m3.large driver, with the following command launching the interactive session:</p>
  
  <pre><code>IPYTHON=1 pyspark --executor-memory 10G --driver-memory 5G --conf spark.driver.maxResultSize=5g
  </code></pre>
  
  <p>In an RDD, if I persist a reference to this broadcast variable, the memory usage explodes. For 100 references to a 100 MB variable, even if it were copied 100 times, I'd expect the data usage to be no more than 10 GB total (let alone 30 GB over 3 nodes). However, I see out of memory errors when I run the following test:</p>
  
  <pre><code>data = list(range(int(10*1e6)))
  metadata = sc.broadcast(data)
  ids = sc.parallelize(zip(range(100), range(100)))
  joined_rdd = ids.mapValues(lambda _: metadata.value)
  joined_rdd.persist()
  print('count: {}'.format(joined_rdd.count()))
  </code></pre>
  
  <p>The stack trace:</p>
  
  <pre><code>TaskSetManager: Lost task 17.3 in stage 0.0 (TID 75, 10.22.10.13): 
  
  org.apache.spark.api.python.PythonException: Traceback (most recent call last):
    File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/worker.py", line 111, in main
      process()
    File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/worker.py", line 106, in process
      serializer.dump_stream(func(split_index, iterator), outfile)
    File "/usr/lib/spark/python/pyspark/rdd.py", line 2355, in pipeline_func
      return func(split, prev_func(split, iterator))
    File "/usr/lib/spark/python/pyspark/rdd.py", line 2355, in pipeline_func
      return func(split, prev_func(split, iterator))
    File "/usr/lib/spark/python/pyspark/rdd.py", line 317, in func
      return f(iterator)
    File "/usr/lib/spark/python/pyspark/rdd.py", line 1006, in &lt;lambda&gt;
      return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()
    File "/usr/lib/spark/python/pyspark/rdd.py", line 1006, in &lt;genexpr&gt;
      return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()
    File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 139, in load_stream
      yield self._read_with_length(stream)
    File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 164, in _read_with_length
      return self.loads(obj)
    File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/serializers.py", line 422, in loads
      return pickle.loads(obj)
  MemoryError
  
  
    at org.apache.spark.api.python.PythonRDD$$anon$1.read(PythonRDD.scala:138)
    at org.apache.spark.api.python.PythonRDD$$anon$1.&lt;init&gt;(PythonRDD.scala:179)
    at org.apache.spark.api.python.PythonRDD.compute(PythonRDD.scala:97)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
    at org.apache.spark.scheduler.Task.run(Task.scala:88)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at org.apache.spark.scheduler.Task.run(Task.scala:88)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
    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)
  
  16/05/25 23:57:15 ERROR TaskSetManager: Task 17 in stage 0.0 failed 4 times; aborting job
  ---------------------------------------------------------------------------
  Py4JJavaError                             Traceback (most recent call last)
  &lt;ipython-input-1-7a262fdfa561&gt; in &lt;module&gt;()
        7 joined_rdd.persist()
        8 print('persist called')
  ----&gt; 9 print('count: {}'.format(joined_rdd.count()))
  
  /usr/lib/spark/python/pyspark/rdd.py in count(self)
     1004         3
     1005         """
  -&gt; 1006         return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()
     1007
     1008     def stats(self):
  
  /usr/lib/spark/python/pyspark/rdd.py in sum(self)
      995         6.0
      996         """
  --&gt; 997         return self.mapPartitions(lambda x: [sum(x)]).fold(0, operator.add)
      998
      999     def count(self):
  
  /usr/lib/spark/python/pyspark/rdd.py in fold(self, zeroValue, op)
      869         # zeroValue provided to each partition is unique from the one provided
      870         # to the final reduce call
  --&gt; 871         vals = self.mapPartitions(func).collect()
      872         return reduce(op, vals, zeroValue)
      873
  
  /usr/lib/spark/python/pyspark/rdd.py in collect(self)
      771         """
      772         with SCCallSiteSync(self.context) as css:
  --&gt; 773             port = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())
      774         return list(_load_from_socket(port, self._jrdd_deserializer))
      775
  
  /usr/lib/spark/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py in __call__(self, *args)
  
    at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:379)
    at py4j.Gateway.invoke(Gateway.java:259)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:207)
    at java.lang.Thread.run(Thread.java:745)
  </code></pre>
  
  <p>I've seen previous threads about the memory usage of pickle deserialization being an issue. However, I would expect a broadcast variable to only be deserialized (and loaded into memory on an executor) once, and subsequent references to <code>.value</code> to reference that in-memory address. That doesn't seem to be the case, however. Am I missing something?</p>
  
  <p>The examples I've seen with broadcast variables have them as dictionaries, used one time to transform a set of data (i.e. replace airport acronyms with airport names). The motivation behind persisting them here is to create objects with knowledge of a broadcast variable and how to interact with it, persist those objects, and perform multiple computations using them (with spark taking care of holding them in memory).</p>
  
  <p>What are some tips for using large (100 MB+) broadcast variables? Is persisting a broadcast variable misguided? Is this an issue that is possibly specific to PySpark?</p>
  
  <p>Thank you! Your help is appreciated.</p>
  
  <p>Note, I've also posted this question on the <a href="https://forums.databricks.com/questions/8371/tips-for-properly-using-large-broadcast-variables.html" rel="noreferrer">databricks forums</a></p>
  
  <p><strong>Edit - followup question:</strong></p>
  
  <p>It was suggested that the default Spark serializer has a batch size of 65337. Objects serialized in different batches are not identified as the same and are assigned different memory addresses, examined here via the builtin <code>id</code> function. However, even with a larger broadcast variable that would in theory take 256 batches to serialize, I still see only 2 distinct copies. Shouldn't I see many more? Is my understanding of how batch serialization works incorrect?</p>
  
  <pre><code>&gt;&gt;&gt; sc.serializer.bestSize
  65536
  &gt;&gt;&gt; import cPickle as pickle
  &gt;&gt;&gt; broadcast_data = {k: v for (k, v) in enumerate(range(int(1e6)))}
  &gt;&gt;&gt; len(pickle.dumps(broadcast_data))
  16777786
  &gt;&gt;&gt; len(pickle.dumps({k: v for (k, v) in enumerate(range(int(1e6)))})) / sc.serializer.bestSize
  256
  &gt;&gt;&gt; bd = sc.broadcast(broadcast_data)
  &gt;&gt;&gt; rdd = sc.parallelize(range(100), 1).map(lambda _: bd.value)
  &gt;&gt;&gt; rdd.map(id).distinct().count()
  1
  &gt;&gt;&gt; rdd.cache().count()
  100
  &gt;&gt;&gt; rdd.map(id).distinct().count()
  2
  </code></pre>