title: Create and save Sequence<Text,Byte[]> files in a foreachPartition function in Spark
body: <p>I am trying to load a group of files, make some checks on them and later saving them in HDFS. I haven't found a good way to create and save these Sequence files, though. Here is my loader main function </p>
  
  <pre><code>SparkConf sparkConf = new SparkConf().setAppName("writingHDFS")
              .setMaster("local[2]")
              .set("spark.streaming.stopGracefullyOnShutdown", "true");
      JavaSparkContext jsc = new JavaSparkContext(sparkConf);
      //JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(5*1000));
  
      JavaPairRDD&lt;String, PortableDataStream&gt; imageByteRDD = jsc.binaryFiles("file:///home/cloudera/Pictures/cat");
      JavaPairRDD&lt;String, String&gt; imageRDD = jsc.wholeTextFiles("file:///home/cloudera/Pictures/");
  
      imageRDD.mapToPair(new PairFunction&lt;Tuple2&lt;String,String&gt;, Text, Text&gt;() {
  
          @Override
          public Tuple2&lt;Text, Text&gt; call(Tuple2&lt;String, String&gt; arg0)
                  throws Exception {
              return new Tuple2&lt;Text, Text&gt;(new Text(arg0._1),new Text(arg0._2));
          }
  
      }).saveAsNewAPIHadoopFile("hdfs://localhost:8020/user/hdfs/sparkling/try.seq", Text.class, Text.class, SequenceFileOutputFormat.class);  
  </code></pre>
  
  <p>It simply loads some images as text files, puts the name of the file as key of the PairRDD and use the native <code>saveAsNewAPIHadoopFile.<br>
  I would like now to save file by file in a</code>rdd.foreach or rdd.foreachPartition` but I cannot find a proper method:  </p>
  
  <ul>
  <li><a href="https://stackoverflow.com/questions/5377118/how-to-convert-txt-file-to-hadoops-sequence-file-format">This stackoverflow answer</a> creates a Job for the occasion. It seems to work, but it needs the file inputed as a path, while I already have an RDD made of them</li>
  <li>A couple of solution I found create a directory for each file (<code>OutputStream out = fs.create(new Path(dst));</code>) which wouldn't be as much of a problem, if it weren't for the fact that I get an exception for <code>Mkdirs didn't work</code></li>
  </ul>
  
  <p>EDIT:  I may have found a way, but I have a <code>Task not serializable</code> exception:<br>
  JavaPairRDD imageByteRDD = jsc.binaryFiles("file:///home/cloudera/Pictures/cat");</p>
  
  <pre><code>    imageByteRDD.foreach(new VoidFunction&lt;Tuple2&lt;String,PortableDataStream&gt;&gt;() {
  
          @Override
          public void call(Tuple2&lt;String, PortableDataStream&gt; fileTuple) throws Exception {
              Text key = new Text(fileTuple._1());
                BytesWritable value = new BytesWritable( fileTuple._2().toArray());
                   SequenceFile.Writer writer = SequenceFile.createWriter(serializableConfiguration.getConf(), SequenceFile.Writer.file(new Path("/user/hdfs/sparkling/" + key)),
                         SequenceFile.Writer.compression(SequenceFile.CompressionType.RECORD, new BZip2Codec()),
                         SequenceFile.Writer.keyClass(Text.class), SequenceFile.Writer.valueClass(BytesWritable.class));
                      key = new Text("MiaoMiao!");
                      writer.append(key, value);
                   IOUtils.closeStream(writer);
  
  
          }
      });
  </code></pre>
  
  <p>I have tried wrapping the entire function in a Serializable class, but no luck. Help?</p>