title: Unable to Iterate over the list of keys retrieved from coverting Dstream to List while using spark streaming with kafka
body: <p>Below is the code for spark streaming with kafka.
  Here I am trying to get the keys for the batch as Dstream and then covert it to a LIST. In order to iterate over it and put data pertaining to each key in a hdfs folder named after the key.</p>
  
  <p>Key is basically  - Schema.Table_name</p>
  
  <pre><code>val ssc = new StreamingContext(sparkConf, Seconds(args{7}.toLong)) // configured to run for every 60 seconds
  val warehouseLocation="Spark-warehouse"
  val spark = SparkSession.builder.config(sparkConf).getOrCreate() 
  import spark.implicits._
  
  val kafkaParams = Map[String, Object](
    "bootstrap.servers" -&gt; conf.getString("kafka.brokers"),
    "zookeeper.connect" -&gt; conf.getString("kafka.zookeeper"),
    "group.id" -&gt; conf.getString("kafka.consumergroups"),
    "auto.offset.reset" -&gt; args { 1 },
    "enable.auto.commit" -&gt; (conf.getString("kafka.autoCommit").toBoolean: java.lang.Boolean),
    "key.deserializer" -&gt; classOf[StringDeserializer],
    "value.deserializer" -&gt; classOf[StringDeserializer],
    "security.protocol" -&gt; "SASL_PLAINTEXT",
    "session.timeout.ms" -&gt; args { 2 },
    "max.poll.records" -&gt; args { 3 },
    "request.timeout.ms" -&gt; args { 4 },
    "fetch.max.wait.ms" -&gt; args { 5 })
  
  val messages = KafkaUtils.createDirectStream[String, String](
    ssc,
    LocationStrategies.PreferConsistent,
    ConsumerStrategies.
    Subscribe[String, String](topicsSet, kafkaParams))
  </code></pre>
  
  <p>Extracting the keys but it is of type DStream[String]</p>
  
  <pre><code> val keys = messages.map(x=&gt;(x.key()))
  
  var final_list_of_keys = List[String]()
  </code></pre>
  
  <p>Converting it into a list and updating var final_list_of_keys</p>
  
  <pre><code>keys.foreachRDD( rdd =&gt; {
  
  val  df_keys = spark.read.json(rdd).distinct().toDF().persist(StorageLevel.MEMORY_ONLY)
  df_keys.show()
  val comma_separated_keys= df_keys.distinct().collect().mkString("").replace("[","").replace("]",",")
  
  final_list_of_keys= comma_separated_keys.split(",").toList
  </code></pre>
  
  <p>Now trying to iterate over the list.</p>
  
  <pre><code> for ( i &lt;- final_list_of_keys)
   {
    println(i)
  
  val message1 = messages.filter(x =&gt;  x.key().toString().equals(i)).map(x=&gt;x.value()).persist(StorageLevel.MEMORY_ONLY) //.toString())
  
   message1.foreachRDD((rdd, batchTime) =&gt; {
  
   if (!rdd.isEmpty())
   {
  
  
     val df1 = spark.read.json(rdd).persist(StorageLevel.MEMORY_ONLY)  //.withColumn("pharmacy_location",lit(args{6}))
  
     val df2=df1.withColumn("message",struct( struct($"message.data.*",lit(args{6}).as("pharmacy_location")).alias("data"), struct($"message.headers.*").as("headers"))).persist(StorageLevel.MEMORY_ONLY)
  
     val df3= df2.drop("headers").drop("messageSchema").drop("messageSchemaId").persist(StorageLevel.MEMORY_ONLY)
  
     df3.coalesce(1).write.json(conf.getString("hdfs.streamoutpath1")+ PATH_SEPERATOR + i + PATH_SEPERATOR + args{6}+ PATH_SEPERATOR+ date_today.format(System.currentTimeMillis())
          + PATH_SEPERATOR + date_today_hour.format(System.currentTimeMillis()) + PATH_SEPERATOR + System.currentTimeMillis())
  
     df1.unpersist
     df2.unpersist()
     df3.unpersist()
  
   }
  
  
  
  })
  
  try
  {
  messages.foreachRDD { rdd =&gt;
    val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
    messages.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)                            // push it back 
  }
  }
  catch
  {
    case e: BlockMissingException =&gt; e.printStackTrace()
   case e: IOException =&gt; e.printStackTrace()
   case e:Throwable =&gt; e.printStackTrace()
  }
  
  }
   ssc.start()
   ssc.awaitTermination()
  </code></pre>
  
  <p>But I get the error - Adding new inputs, transformations, and output operations after starting a context is not supported</p>
  
  <p>When I tried to keep the for loop on list outside the keys.foreachRdd then the list does not get updated and remains empty.</p>
  
  <p>Can someone please advice how can I actually redo this code to have the keys in a list then go over them to put data in correct directory.</p>
  
  <p>From my research i saw post -</p>
  
  <p><a href="https://stackoverflow.com/questions/36132844/java-lang-illegalstateexception-adding-new-inputs-transformations-and-output">Similar post but unable to gather any solution from it</a></p>
  
  <p>Also,as I am using map,filter inside foreachRdd and then another foreachRdd inside it can cause a problem.
  Refer post - <a href="https://stackoverflow.com/questions/37303202/about-an-error-accessing-a-field-inside-tuple2">Post with similar code</a>      </p>