如何执行一对多映射缩减联接

Question

当我们有以下情况时，一对多连接情况呢：

文件1

人员 ID1，名称 1

人员 ID2，名称 2

文件2

人员 ID1，地址 2

文件2

人员 ID2，地址 2

我想有减速机输出

人员 ID1、名称 1、地址 2

人员 ID2、名称 2、地址 2

Answer 1

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

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

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);
  }
}

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

Answer 2

我假设每个personid只能有一个名称，但可以有多个地址。

映射器应扫描所有输入文件并生成如下所示的键值对：

(personid1, (0, name1))
(personid2, (0, name2))
(personid1, (1, address2))
(personid2, (1, address2))

整数标志0表示记录来自file1，标志1表示记录来自其他类型的文件。

减速器输入为：

(personid1, [(0, name1), (1, address2)])
(personid2, [(1, address2), (0, name2)])

Hadoop 无法保证 shuffle 输出中原始值的顺序，所以我在上面的第二行中更改了这个顺序，只是为了说明这一点。化简器的工作是解码每条记录的值（方括号中的列表） - 与flag = 0 (flag, value)的对将为您提供名称，所有其他对将为您提供此人的所有地址。

享受Hadoop！