Spark - 将RDD保存到多个文件作为输出

Question

我有一个JavaRDD<Model>，我需要将它写成多个具有不同布局的文件[RDD 中的一个或两个字段在不同布局之间会有所不同]。

当我使用saveAsTextFile()它调用模型的toString()方法时，这意味着相同的布局将被写入输出。

目前我正在做的是使用 map 转换方法迭代 RDD 并返回具有其他布局的不同模型，因此我可以使用 saveAsTextFile() 操作写入不同的输出文件。

由于一两个字段不同，我需要再次迭代整个RDD并创建新的RDD然后将其保存为输出文件。

例如：

具有字段的当前 RDD：

RoleIndicator、姓名、年龄、地址、部门

输出文件 1：

姓名、年龄、地址

输出文件 2：

RoleIndicator、姓名、年龄、部门

有没有最佳解决方案？

问候， 尚卡尔

Answer 1

您想使用foreach，而不是collect。

您应该将您的函数定义为扩展VoidFunction 的实际命名类。为这两个文件创建实例变量，并添加一个关闭文件的close() 方法。您的 call() 实现将编写您需要的任何内容。

完成后记得在你的函数对象上调用close()。

Answer 2

可以使用 Pair RDD。 使用 Hadoop 自定义输出格式，Pair RDD 可以在一次迭代中存储在多个文件中。

rdd.saveAsHadoopFile(path, key.class, value.class,CustomTextOutputFormat.class, jobConf);


public class FileGroupingTextOutputFormat extends MultipleTextOutputFormat<Text, Text> { 
  @Override
  protected Text generateActualKey(Text key, Text value) {
    return new Text();
  }

  @Override
  protected Text generateActualValue(Text key, Text value) {
    return value;
  }
// returns a dynamic file name based on each RDD element
  @Override
  protected String generateFileNameForKeyValue(Text key, Text value, String name) {
    return value.getSomeField() + "-" + name;
  }
}