title: Convert each record in RDD to a Array[Map] using scala and Spark
body: <p>My RDD is  \n separated  records that look like</p>
  
  <pre><code>Single RDD
  
  k1=v1,k2=v2,k3=v3
  k1=v1,k2=v2,k3=v3
  k1=v1,k2=v2,k3=v3
  </code></pre>
  
  <p>and want to convert it into a Array[Map[k,v]], <br/> <br/>
  where each element in Array will be a different map[k,v] corresponding to a record.  <br/> <br/>
  Array will contain N number of such maps depending on the records in a single RDD.</p>
  
  <p>I am new to both scala and spark. Any help in the conversion will help. </p>
  
  <pre><code>object SparkApp  extends Logging with App {
  
  
    override def main(args: Array[ String ]): Unit = {
      val myConfigFile = new File("../sparkconsumer/conf/spark.conf")
      val fileConfig = ConfigFactory.parseFile(myConfigFile).getConfig(GlobalConstants.CONFIG_ROOT_ELEMENT)
      val propConf = ConfigFactory.load(fileConfig)
      val topicsSet = propConf.getString(GlobalConstants.KAFKA_WHITE_LIST_TOPIC).split(",").toSet
      val kafkaParams = Map[ String, String ]("metadata.broker.list" -&gt; propConf.getString(GlobalConstants.KAFKA_BROKERS))
  
  
      //logger.info(message = "Hello World , You are entering Spark!!!")
      val conf = new SparkConf().setMaster("local[2]").setAppName(propConf.getString(GlobalConstants.JOB_NAME))
      conf.set("HADOOP_HOME", "/usr/local/hadoop")
      conf.set("hadoop.home.dir", "/usr/local/hadoop")
      //Lookup
  
      // logger.info("Window of 5 Seconds Enabled")
      val ssc = new StreamingContext(conf, Seconds(5))
      ssc.checkpoint("/tmp/checkpoint")
  
      val apiFile = ssc.sparkContext.textFile(propConf.getString(GlobalConstants.API_FILE))
      val arrayApi = ssc.sparkContext.broadcast(apiFile.distinct().collect())
  
      val nonApiFile = ssc.sparkContext.textFile(propConf.getString(GlobalConstants.NON_API_FILE))
      val arrayNonApi = ssc.sparkContext.broadcast(nonApiFile.distinct().collect())
  
  
      val messages = KafkaUtils.createDirectStream[ String, String, StringDecoder, StringDecoder ](ssc, kafkaParams, topicsSet)
      writeTOHDFS2(messages)
      ssc.start()
      ssc.awaitTermination()
    }
  
  
  
    def writeTOHDFS2(messages: DStream[ (String, String) ]): Unit = {
      val records = messages.window(Seconds(10), Seconds(10))
      val k = records.transform( rdd =&gt; rdd.map(r =&gt;r._2)).filter(x=&gt; filterNullImpressions(x))
  
      k.foreachRDD { singleRdd =&gt;
        if (singleRdd.count() &gt; 0) {
  
  
          val maps =  singleRdd.map(line =&gt; line.split("\n").flatMap(x =&gt; x.split(",")).flatMap(x =&gt; x.split("=")).foreach( x =&gt; new mutable.HashMap().put(x(0),x(1))) 
  
  
          val r = scala.util.Random
          val sdf = new SimpleDateFormat("yyyy/MM/dd/HH/mm")
          maps.saveAsTextFile("hdfs://localhost:8001/user/hadoop/spark/" + sdf.format(new Date())+r.nextInt)
        }
      }
  
    }
  
  }
  </code></pre>