hadoop - 如何执行一对多 map-reduce 连接？

Question

当我们有:

文件1

personid1, name1

personid2, name2

文件2

personid1, address2

文件2

personid2, address2

我想要reducer输出

personid1, name1, address2

personid2, name2, address2

Answer 1

您似乎可以使用 personid 作为映射器的键。

然后您将确保在一个 reducer 中获得属于一个 personid 的所有记录作为迭代器。现在您需要区分哪个记录来自哪个来源，因此最好将标识符放在值上。

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class ExampleDriver extends Configured implements Tool {


  @Override
  public int run(String[] args) throws Exception {
    Configuration configuration = getConf();

    Job job = Job.getInstance(configuration, this.getClass().getName());

    job.setJarByClass(ExampleDriver.class);

    MultipleInputs.addInputPath(job, new Path(PERSON_DIR), TextInputFormat.class, PersonMapper.class);
    MultipleInputs.addInputPath(job, new Path(ADDRESS_DIR), TextInputFormat.class, AddressMapper.class);


    job.setMapOutputKeyClass(KeyWithPersonId.class);
    job.setMapOutputValueClass(Text.class);
    job.setReducerClass(JoinPersonWithAddressReducer.class);

    LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class); // Not necessary. Can use simple FileOutputFormat.

    return job.waitForCompletion(true) ? 0 : 1;
  }

  public static void main(String[] args) throws Exception {
    int exitCode = ToolRunner.run(new Configuration(), new ExampleDriver(), args);
    System.exit(exitCode);
  }
}

如果您有更多问题，请告诉我。