title: Spark Streaming mapWithState seems to rebuild complete state periodically
body: <p>I am working on a Scala (2.11) / Spark (1.6.1) streaming project and using <code>mapWithState()</code> to keep track of seen data from previous batches. </p>
  
  <p>The state is distributed in 20 partitions on multiple nodes, created with <code>StateSpec.function(trackStateFunc _).numPartitions(20)</code>. In this state we have only a few keys (~100) mapped to <code>Sets</code> with up ~160.000 entries, which grow throughout the application. The entire state is up to <code>3GB</code>, which can be handled by each node in the cluster. In each batch, some data is added to a state but not deleted until the very end of the process, i.e. ~15 minutes.</p>
  
  <p>While following the application UI, every 10th batch's processing time is very high compared to the other batches. See images:</p>
  
  <p><a href="https://i.stack.imgur.com/BfeoA.png" rel="noreferrer"><img src="https://i.stack.imgur.com/BfeoA.png" alt="The spikes show the higher processing time."></a></p>
  
  <p>The yellow fields represent the high processing time.</p>
  
  <p><a href="https://i.stack.imgur.com/HdpQn.png" rel="noreferrer"><img src="https://i.stack.imgur.com/HdpQn.png" alt="enter image description here"></a></p>
  
  <p>A more detailed Job view shows that in these batches occur at a certain point, exactly when all 20 partitions are "skipped". Or this is what the UI says.</p>
  
  <p><a href="https://i.stack.imgur.com/OLeQX.png" rel="noreferrer"><img src="https://i.stack.imgur.com/OLeQX.png" alt="enter image description here"></a></p>
  
  <p>My understanding of <code>skipped</code> is that each state partition is one possible task which isn't executed, as it doesn't need to be recomputed. However, I don't understand why the amount of <code>skips</code> varies in each Job and why the last Job requires so much processing. The higher processing time occurs regardless of the state's size, it just impacts the duration.</p>
  
  <p>Is this a bug in the <code>mapWithState()</code> functionality or is this intended behaviour? Does the underlying data structure require some kind of reshuffling, does the <code>Set</code> in the state need to copy data? Or is it more likely to be a flaw in my application?</p>