在Hadoop中将减少的数据拆分为输出和新输入

Question

我一直在寻找使用简化数据的方法,以便在hadoop中进一步映射.我有类的对象A作为输入数据和类的对象B作为输出数据.问题是,虽然映射不仅B生成s而且生成新As.

这就是我想要实现的目标:

1.1 input: a list of As
1.2 map result: for each A a list of new As and a list of Bs is generated
1.3 reduce: filtered Bs are saved as output, filtered As are added to the map jobs

2.1 input: a list of As produced by the first map/reduce
2.2 map result: for each A a list of new As and a list of Bs is generated
2.3 ...

3.1 ...

你应该得到基本的想法.

我已经阅读了很多关于链接的内容,但我不确定如何将ChainReducer和ChainMapper结合起来,或者即使这是正确的方法.

所以这是我的问题:如何在减少时拆分映射数据,将一个部分保存为输出,另一部分保存为新的输入数据.

Answer 1

尝试使用MultipleOutputs。正如 Javadoc 所建议的：

MultipleOutputs 类简化了将输出数据写入多个输出的过程

情况一：写入作业默认输出之外的其他输出。每个附加输出或命名输出都可以使用其自己的 OutputFormat、其自己的键类和其自己的值类进行配置。

情况二：将数据写入用户提供的不同文件中

作业提交的使用模式：

Job job = new Job();

 FileInputFormat.setInputPath(job, inDir);
 FileOutputFormat.setOutputPath(job, outDir);

 job.setMapperClass(MOMap.class);
 job.setReducerClass(MOReduce.class);
 ...

 // Defines additional single text based output 'text' for the job
 MultipleOutputs.addNamedOutput(job, "text", TextOutputFormat.class,
 LongWritable.class, Text.class);

 // Defines additional sequence-file based output 'sequence' for the job
 MultipleOutputs.addNamedOutput(job, "seq",
   SequenceFileOutputFormat.class,
   LongWritable.class, Text.class);
 ...

 job.waitForCompletion(true);
 ...

在减速器中的用法：

 String generateFileName(K k, V v) {
   return k.toString() + "_" + v.toString();
 }

 public class MOReduce extends
   Reducer<WritableComparable, Writable,WritableComparable, Writable> {
 private MultipleOutputs mos;
 public void setup(Context context) {
 ...
 mos = new MultipleOutputs(context);
 }

 public void reduce(WritableComparable key, Iterator<Writable> values,
 Context context)
 throws IOException {
 ...
 mos.write("text", , key, new Text("Hello"));
 mos.write("seq", LongWritable(1), new Text("Bye"), "seq_a");
 mos.write("seq", LongWritable(2), key, new Text("Chau"), "seq_b");
 mos.write(key, new Text("value"), generateFileName(key, new Text("value")));
 ...
 }

 public void cleanup(Context) throws IOException {
 mos.close();
 ...
 }

 }