title: Optimal way of creating a cache in the PySpark environment
body: <p>I am using Spark Streaming for creating a system to enrich incoming data from a cloudant database. Example - </p>
  
  <pre><code>Incoming Message: {"id" : 123}
  Outgoing Message: {"id" : 123, "data": "xxxxxxxxxxxxxxxxxxx"}
  </code></pre>
  
  <p>My code for the driver class is as follows:</p>
  
  <pre class="lang-py prettyprint-override"><code>from Sample.Job import EnrichmentJob
  from Sample.Job import FunctionJob
  import pyspark
  from pyspark.streaming.kafka import KafkaUtils
  from pyspark import SparkContext, SparkConf, SQLContext
  from pyspark.streaming import StreamingContext
  from pyspark.sql import SparkSession
  
  from kafka import KafkaConsumer, KafkaProducer
  import json
  
  class SampleFramework():
  
      def __init__(self):
          pass
  
      @staticmethod
      def messageHandler(m):
          return json.loads(m.message)
  
      @staticmethod
      def processData(rdd):
  
          if (rdd.isEmpty()):
              print("RDD is Empty")
              return
  
          # Expand
          expanded_rdd = rdd.mapPartitions(EnrichmentJob.enrich)
  
          # Score
          scored_rdd = expanded_rdd.map(FunctionJob.function)
  
          # Publish RDD
  
  
      def run(self, ssc):
  
          self.ssc = ssc
  
          directKafkaStream = KafkaUtils.createDirectStream(self.ssc, QUEUENAME, \
                                                            {"metadata.broker.list": META, 
                                                            "bootstrap.servers": SERVER}, \
                                                            messageHandler= SampleFramework.messageHandler)
  
          directKafkaStream.foreachRDD(SampleFramework.processData)
  
          ssc.start()
          ssc.awaitTermination()
  </code></pre>
  
  <p>Code for the the Enrichment Job is as follows:
  class EnrichmentJob:</p>
  
  <pre class="lang-py prettyprint-override"><code>cache = {}
  
  @staticmethod
  def enrich(data):
  
      # Assume that Cloudant Connector using the available config
      cloudantConnector = CloudantConnector(config, config["cloudant"]["host"]["req_db_name"])
      final_data = []
      for row in data:
          id = row["id"]
          if(id not in EnrichmentJob.cache.keys()):
              data = cloudantConnector.getOne({"id": id})
              row["data"] = data
              EnrichmentJob.cache[id]=data
          else:
              data = EnrichmentJob.cache[id]
              row["data"] = data
          final_data.append(row)
  
      cloudantConnector.close()
  
      return final_data
  </code></pre>
  
  <p>My question is - Is there someway to maintain [1]"a global cache on the main memory that is accessible to all workers" or [2]"local caches on each of the workers such that they remain persisted in the foreachRDD setting"? </p>
  
  <p>I have already explored the following -</p>
  
  <ol>
  <li><p>Broadcast Variables - Here we go the [1] way. As I understand, they are meant to be read-only and immutable. I have checked out this <a href="https://stackoverflow.com/questions/39729059/spark-streaming-and-mutable-broadcast-variable?rq=1">reference</a> but it cites an example of unpersisting/persisting the broadcasted variable. Is this a good practice?</p></li>
  <li><p>Static Variables - Here we go the [2] way. The class that is being referred to ("Enricher" in this case) maintains a cache in the form of a static variable dictionary. But it turns out that the ForEachRDD function spawns a completely new process for each incoming RDD and this removes the previously initiated static variable. This is the one coded above.</p></li>
  </ol>
  
  <p>I have two possible solutions right now -</p>
  
  <ol>
  <li>Maintain an offline cache on the file system. </li>
  <li>Do the entire computation of this enrichment task on my driver node. This would cause the entire data to end up on driver and be maintained there. The cache object will be sent to the enrichment job as an argument to the mapping function. </li>
  </ol>
  
  <p>Here obviously the first one looks better than the second, but I wish to conclude that these two are the only ways around, before committing to them. Any pointers would be appreciated!</p>