title: RDD[(String,Iterable[GenericData.Record])] to Map[(String,RDD[GenericData.Record])]
body: <p>I have an <code>RDD</code> which is of type <code>(String,Iterable[GenericData.Record])</code>. Now I want to save these iterables to paths based on the keys of this RDD. So for example if the RDD contained</p>
  <pre><code>(&quot;a&quot;,[1,2,3,4])
  (&quot;b&quot;,[5,6,7,9])
  </code></pre>
  <p>I need to perist [1,2,3,4] under <code>result-path/a</code> and [5,6,7,8,9] under <code>result-path/b</code>. One way to do this - which compiles but fails on runtime - is the following:</p>
  <pre class="lang-scala prettyprint-override"><code>implicit val spark: SparkSession = SparkSessionUtils.initSparkSession(&quot;Test&quot;)
  implicit val sc: SparkContext = spark.sparkContext
  val re:RDD[(String,Iterable[GenericData.Record])] = ???
  val hadoopConf = new Configuration(sc.hadoopConfiguration)
  
  re.forearch {
     case (key,collection) =&gt;
         val reRDD = sc.makeRDD(collection)
         reRDD.saveAsNewAPIHadoopFile(s&quot;$uri/$key&quot;,
          classOf[SpecificRecord],
          classOf[NullWritable],
          classOf[AvroKeyOutputFormat[SpecificRecord]],
          hadoopConf)
  
  }
  </code></pre>
  <p>The problem here is that I cannot do this since SparkContext is not serializable.<code>re</code> is an <code>RDD</code> so invoking <code>foreach</code> on it must serialize the inner lambda and send over to the worker nodes. So I am trying to think of a way where I can convert the initial <code>re</code> into a <code>Map[(String,RDD[GenericData.Record])]</code> so that I can do the following:</p>
  <pre class="lang-scala prettyprint-override"><code>implicit val spark: SparkSession = SparkSessionUtils.initSparkSession(&quot;Test&quot;)
  implicit val sc: SparkContext = spark.sparkContext
  val re:Map[(String,RDD[GenericData.Record])] = ???
  val hadoopConf = new Configuration(sc.hadoopConfiguration)
  
  re.forearch {
     case (key,rddCollection) =&gt;
         rddCollection.saveAsNewAPIHadoopFile(s&quot;$uri/$key&quot;,
          classOf[SpecificRecord],
          classOf[NullWritable],
          classOf[AvroKeyOutputFormat[SpecificRecord]],
          hadoopConf)
  
  }
  </code></pre>