Spark输出自定义文件目录踩坑(Java)

2,415 阅读4分钟

最近项目中,使用Spark做离线计算,结果需要输出一份结果到文件中保存,并且需要按Key来放置不同的目录。因为spark通过saveAsTextFile()方法默认输出是以part-0000的形式。

解决方法

通过搜索,很轻易的就能搜索到使用saveAsHadoopFile()方法可以将文件输出到自定义文件目录。网上大部分都是scala的写法,java的具体操作如下:

//首先,构造出一个PariRDD形式的RDD
JavaPairRDD<String, JSONObject> javaPairRDD =xxx
//使用saveAsHadoopFile方法输出到目标目录下,方法参数分别为(目标目录,key的class类型,value的class类型,输出format类)
javaPairRDD.saveAsHadoopFile("D:\\Test",String.class,JSONObject.class,RDDMultipleTextOutputFormat2.class);

//自定义一个RDDMultipleTextOutputFormat2继承MultipleTextOutputFormat
public static class RDDMultipleTextOutputFormat2 extends MultipleTextOutputFormat<String, JSONObject> {

    @Override
    public String generateFileNameForKeyValue(String key, JSONObject value,
                                              String name) {
        String object_type = value.getString("object_type");
        String object_id = value.getString("object_id");
        return object_type + "/" + object_id+".json";
    }

}

最后输出的结果就是按"D:\Test\object_type\object_id.json来区分保存。

新的问题

美滋滋的打开按我们要求输出目录的输出文件。结果却发现,输出文件中,并不是仅仅将value写入了文件中,同时把key也写了进去。但是我们的文件格式是不需要key写入文件的。

//输出文件内容形式如下
key  value

解决办法一

遇事不决百度Google,通过搜索引擎,可以找到我们通过设置rdd的key为NullWritable,使得输出文件中不包含key,网上同样大多数是scala的,下面是java的具体操作:

/首先,构造出一个PariRDD形式的RDD
JavaPairRDD<String, JSONObject> javaPairRDD =xxx

//将PariRDD转为<NullWritable,T>的形式
JavaPairRDD<NullWritable, JSONObject> nullKeyJavaPairRDD = javaPairRDD.mapToPair(tuple2 -> {
    return new Tuple2(NullWritable.get(),tuple2._2);
});

//接下来的操作和上面一样
nullKeyJavaPairRDD.saveAsHadoopFile("D:\\Test",NullWritable.class,JSONObject.class,RDDMultipleTextOutputFormat.class);

public static class RDDMultipleTextOutputFormat extends MultipleTextOutputFormat<NullWritable, JSONObject> {

    @Override
    public String generateFileNameForKeyValue(NullWritable key, JSONObject value,
                                              String name) {
        String object_type = value.getString("object_type");
        String object_id = value.getString("object_id");
        return object_type + "/" + object_id+".json";
    }
}

如上方法确实是可以解决问题,但是如果我们需要的输出目录与key有关系,想将key使用在自定义目录中,这就办不到了。所以这个解决方法还是有缺陷的,仅仅能满足输出目录只与value有关系的情况。

解决办法二

这里想到另一个解决思路,往文件写内容总是需要一个类的,我们找到这个类,重写它,把key的输出去掉,不就可以了。

于是我们跟进我们继承的MultipleTextOutputFormat类中

public class MultipleTextOutputFormat<K, V>
    extends MultipleOutputFormat<K, V> {

  private TextOutputFormat<K, V> theTextOutputFormat = null;

  @Override
  protected RecordWriter<K, V> getBaseRecordWriter(FileSystem fs, JobConf job,
      String name, Progressable arg3) throws IOException {
    if (theTextOutputFormat == null) {
      theTextOutputFormat = new TextOutputFormat<K, V>();
    }
    return theTextOutputFormat.getRecordWriter(fs, job, name, arg3);
  }
}

并没有发现有相关的方法,我们继续跟进父类MultipleOutputFormat,在这个类中,我们发现了一个write方法:

public void write(K key, V value) throws IOException {

        // get the file name based on the key
        String keyBasedPath = generateFileNameForKeyValue(key, value, myName);

        // get the file name based on the input file name
        String finalPath = getInputFileBasedOutputFileName(myJob, keyBasedPath);

        // get the actual key
        K actualKey = generateActualKey(key, value);
        V actualValue = generateActualValue(key, value);

        RecordWriter<K, V> rw = this.recordWriters.get(finalPath);
        if (rw == null) {
          // if we don't have the record writer yet for the final path, create
          // one
          // and add it to the cache
          rw = getBaseRecordWriter(myFS, myJob, finalPath, myProgressable);
          this.recordWriters.put(finalPath, rw);
        }
        rw.write(actualKey, actualValue);
      };

感觉真相就在眼前了,我们继续跟进rw.write(actualKey, actualValue);方法,通过断点我们可以知道他进入的是TextOutPutFormat #write()方法:

public synchronized void write(K key, V value)
      throws IOException {

      boolean nullKey = key == null || key instanceof NullWritable;
      boolean nullValue = value == null || value instanceof NullWritable;
      if (nullKey && nullValue) {
        return;
      }
      if (!nullKey) {
        writeObject(key);
      }
      if (!(nullKey || nullValue)) {
        out.write(keyValueSeparator);
      }
      if (!nullValue) {
        writeObject(value);
      }
      out.write(newline);
    }

这段代码就很简单了,只要key不是null或者NullWritable类,他就会往文件里输出。这也解释了为什么上面方法一中将key转换为NullWritable类就不会输出到文件中了。

了解到这里,我们就很容易得出解决方案,我们只要将传入write()方法中的key传入null就可以了。回到MultipleOutputFormat类中,我们看到传入的key是由这个方法获取的K actualKey = generateActualKey(key, value);,继续跟进:

  protected K generateActualKey(K key, V value) {
    return key;
  }

接下很简单了,我们在自定义的format类中重写这个方法,改为返回null即可。

    public static class RDDMultipleTextOutputFormat3 extends MultipleTextOutputFormat<String, JSONObject> {

        @Override
        public String generateFileNameForKeyValue(String key, JSONObject value,
                                                  String name) {
            String object_id = value.getString("object_id");
            return key + "/" + object_id+".json";
        }
        
        @Override
        public String generateActualKey(String key, JSONObject value) {
            return null;
        }
    }

总结

其实这次踩坑还是挺简单的,按部就班的一路跟进就找到了,其中还有不少点是可以延伸的,比如saveAsHadoopFile方法的的底层实现,和传统的saveAsTextFile方法的异同,我在查资料的过程中也发现了很多延伸的知识点,不过这些应该就是另一篇博客了。