在 Spark 的 foreachPartition 函数中创建并保存 Sequence<Text,Byte[]> 文件

Question

我正在尝试加载一组文件，对它们进行一些检查，然后将它们保存在 HDFS 中。不过，我还没有找到创建和保存这些序列文件的好方法。这是我的加载器主要功能

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<String, PortableDataStream> imageByteRDD = jsc.binaryFiles("file:///home/cloudera/Pictures/cat");
    JavaPairRDD<String, String> imageRDD = jsc.wholeTextFiles("file:///home/cloudera/Pictures/");

    imageRDD.mapToPair(new PairFunction<Tuple2<String,String>, Text, Text>() {

        @Override
        public Tuple2<Text, Text> call(Tuple2<String, String> arg0)
                throws Exception {
            return new Tuple2<Text, Text>(new Text(arg0._1),new Text(arg0._2));
        }

    }).saveAsNewAPIHadoopFile("hdfs://localhost:8020/user/hdfs/sparkling/try.seq", Text.class, Text.class, SequenceFileOutputFormat.class);  

它只是将一些图像加载为文本文件，将文件名作为 PairRDD 的键并使用本机 saveAsNewAPIHadoopFile.
I would like now to save file by file in ardd.foreach 或 rdd.foreachPartition` 但我找不到合适的方法：

• This stackoverflow answer 为该场合创建工作。它似乎工作，但它需要作为路径输入的文件，而我已经有一个由它们组成的 RDD
• 我找到了几个解决方案，为每个文件 (OutputStream out = fs.create(new Path(dst));) 创建一个目录，如果不是因为我得到了 Mkdirs didn't work 的例外情况，这不会有太大问题

编辑：我可能已经找到了方法，但我有一个Task not serializable 异常：
JavaPairRDD imageByteRDD = jsc.binaryFiles("file:///home/cloudera/Pictures/cat");

    imageByteRDD.foreach(new VoidFunction<Tuple2<String,PortableDataStream>>() {

        @Override
        public void call(Tuple2<String, PortableDataStream> 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);


        }
    });

我尝试将整个函数包装在 Serializable 类中，但没有成功。帮忙？

Answer 1

我这样做的方式是（伪代码，我一到办公室就会尝试编辑这个答案）

rdd.foreachPartition{
    Configuration conf = ConfigurationSingletonClass.getConfiguration();
    etcetera, etcetera...
}

编辑：到我的办公室，这里是完整的代码段：配置是在 inside rdd.foreachPartition 中创建的（每个都有点太多了） .在迭代器中，文件将自身写入序列文件格式。

 JavaPairRDD<String, PortableDataStream> imageByteRDD = jsc.binaryFiles(SOURCE_PATH);
    if(!imageByteRDD.isEmpty())
        imageByteRDD.foreachPartition(new VoidFunction<Iterator<Tuple2<String,PortableDataStream>>>() {

            @Override
            public void call(
                    Iterator<Tuple2<String, PortableDataStream>> arg0)
                    throws Exception {
                Configuration conf = new Configuration();
                conf.set("fs.defaultFS", HDFS_PATH);
                while(arg0.hasNext()){
                    Tuple2<String,PortableDataStream>fileTuple = arg0.next();
                    Text key = new Text(fileTuple._1());
                    String fileName = key.toString().split(SEP_PATH)[key.toString().split(SEP_PATH).length-1].split(DOT_REGEX)[0];
                    String fileExtension = fileName.split(DOT_REGEX)[fileName.split(DOT_REGEX).length-1];

                      BytesWritable value = new BytesWritable( fileTuple._2().toArray());
                         SequenceFile.Writer writer = SequenceFile.createWriter(
                                 conf, 
                                 SequenceFile.Writer.file(new Path(DEST_PATH + fileName + SEP_KEY + getCurrentTimeStamp()+DOT+fileExtension)),
                               SequenceFile.Writer.compression(SequenceFile.CompressionType.RECORD, new BZip2Codec()),
                               SequenceFile.Writer.keyClass(Text.class), SequenceFile.Writer.valueClass(BytesWritable.class));
                         key = new Text(key.toString().split(SEP_PATH)[key.toString().split(SEP_PATH).length-2] + SEP_KEY + fileName + SEP_KEY + fileExtension);
                            writer.append(key, value);
                         IOUtils.closeStream(writer);

                }
            }
        });

希望这会有所帮助。