title: Efficient countByValue of each column Spark Streaming
body: <p>I want to find countByValues of each column in my data. I can find countByValue() for each column (e.g. 2 columns now) in basic batch RDD as fallows:</p>
  
  <pre><code>scala&gt; val double = sc.textFile("double.csv") 
  scala&gt; val counts = sc.parallelize((0 to 1).map(index =&gt; {
       double.map(x=&gt; { val token = x.split(",")
       (math.round(token(index).toDouble))
       }).countByValue()
       }))
  scala&gt; counts.take(2)
  res20: Array[scala.collection.Map[Long,Long]] = Array(Map(2 -&gt; 5, 1 -&gt; 5), Map(4 -&gt; 5, 5 -&gt; 5))
  </code></pre>
  
  <p>Now I want to perform same with DStreams. I have windowedDStream and want to countByValue on each column. My data has 50 columns. I have done it as fallows:</p>
  
  <pre><code>val windowedDStream = myDStream.window(Seconds(2), Seconds(2)).cache()  
  ssc.sparkContext.parallelize((0 to 49).map(index=&gt; {
  val counts = windowedDStream.map(x=&gt; { val token = x.split(",")
                                  (math.round(token(index).toDouble))
                                  }).countByValue()
  counts.print()                                          
  }))                                                
  val topCounts = counts.map . . . .  will not work
  </code></pre>
  
  <p>I get correct results with this, the only issue is that I want to apply more operations on counts and it's not available outside map.</p>