title: Spark streaming: Cache DStream results across batches
body: <p>Using Spark streaming (1.6) I have a filestream for reading lookup data with 2s of batch size, however files are copyied to the directory only every hour.<br>
  Once there's a new file, its content is read by the stream, this is what I want to cache into memory and keep there
  until new files are read. <br>
  There's another stream to which I want to join this dataset therefore I'd like to cache.<br></p>
  
  <p>This is a follow-up question of <a href="https://stackoverflow.com/questions/37447393/batch-lookup-data-for-spark-streaming">Batch lookup data for Spark streaming</a>.<br>
  The answer does work fine with <code>updateStateByKey</code> however I don't know how to deal with cases when a KV pair is 
  <em>deleted</em> from the lookup files, as the Sequence of values in <code>updateStateByKey</code> keeps growing.
  Also any hint how to do this with <code>mapWithState</code> would be great.</p>
  
  <p>This is what I tried so far, but the data doesn't seem to be persisted:</p>
  
  <pre><code>val dictionaryStream = ssc.textFileStream("/my/dir")
  dictionaryStream.foreachRDD{x =&gt; 
    if (!x.partitions.isEmpty) {
      x.unpersist(true)
      x.persist()
    }
  }
  </code></pre>