title: Spark column wise word count
body: <p>We are trying to generate column wise statistics of our dataset in spark. In addition to using the summary function from statistics library. We are using the following procedure:</p>
  
  <ol>
  <li><p>We determine the columns with string values</p></li>
  <li><p>Generate key value pair for the whole dataset, using the column number as key and value of column as value</p></li>
  <li><p>generate a new map of format</p>
  
  <p>(K,V) ->((K,V),1)</p></li>
  </ol>
  
  <p>Then we use reduceByKey to find the sum of all unique value in all the columns. We cache this output to reduce further computation time.</p>
  
  <p>In the next step we cycle through the columns using a for loop to find the statistics for all the columns.</p>
  
  <p>We are trying to reduce the for loop by again utilizing the map reduce way but we are unable to find some way to achieve it. Doing so will allow us to generate column statistics for all columns in one execution. The for loop method is running sequentially making it very slow.</p>
  
  <p>Code: </p>
  
  <pre><code>//drops the header
  
      def dropHeader(data: RDD[String]): RDD[String] = {
           data.mapPartitionsWithIndex((idx, lines) =&gt; {
             if (idx == 0) {
               lines.drop(1)
             }
             lines
           })
         }
  
      def retAtrTuple(x: String) = {
         val newX = x.split(",")
         for (h &lt;- 0 until newX.length) 
            yield (h,newX(h))
      }
  
  
  
      val line = sc.textFile("hdfs://.../myfile.csv")
  
      val withoutHeader: RDD[String] = dropHeader(line)
  
      val kvPairs = withoutHeader.flatMap(retAtrTuple) //generates a key-value pair where key is the column number and value is column's value
  
  
      var bool_numeric_col = kvPairs.map{case (x,y) =&gt; (x,isNumeric(y))}.reduceByKey(_&amp;&amp;_).sortByKey()    //this contains column indexes as key and boolean as value (true for numeric and false for string type)
  
      var str_cols = bool_numeric_col.filter{case (x,y) =&gt; y == false}.map{case (x,y) =&gt; x}
      var num_cols = bool_numeric_col.filter{case (x,y) =&gt; y == true}.map{case (x,y) =&gt; x}
  
      var str_col = str_cols.toArray   //array consisting the string col
      var num_col = num_cols.toArray   //array consisting numeric col
  
  
      val colCount = kvPairs.map((_,1)).reduceByKey(_+_)
      val e1 = colCount.map{case ((x,y),z) =&gt; (x,(y,z))}
      var numPairs = e1.filter{case (x,(y,z)) =&gt; str_col.contains(x) }
  
      //running for loops which needs to be parallelized/optimized as it sequentially operates on each column. Idea is to find the top10, bottom10 and number of distinct elements column wise
      for(i &lt;- str_col){
         var total = numPairs.filter{case (x,(y,z)) =&gt; x==i}.sortBy(_._2._2)
         var leastOnes = total.take(10)
         println("leastOnes for Col" + i)
         leastOnes.foreach(println)
         var maxOnes = total.sortBy(-_._2._2).take(10)
         println("maxOnes for Col" + i)
         maxOnes.foreach(println)
         println("distinct for Col" + i + " is " + total.count)
      }
  </code></pre>