如何执行一对多Mapreduce连接?

uemypmqf  于 2021-05-30  发布在  Hadoop
关注(0)|答案(2)|浏览(297)

当我们有以下情况时,一对多连接情况如何:
文件1
人格1,姓名1
人格2,姓名2
文件2
拟人1,地址2
文件2
拟人2,地址2
我想要减速机输出
人格1,姓名1,地址2
拟人2,姓名2,地址2

hjzp0vay

hjzp0vay1#

我假设每个人 personid 只能有一个名称,但可以有多个地址。
Map程序应该扫描所有输入文件并生成如下键值对:

(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输出中原始值的顺序,所以我在上面的第二行中更改了这个顺序来说明这一点。reducer的任务是对每个记录(方括号中的列表)的值进行解码 (flag, value)flag = 0 我会告诉你名字,其他人会告诉你这个人的所有地址。
享受hadoop!

p8h8hvxi

p8h8hvxi2#

似乎你可以使用personid作为Map绘制者的钥匙。
然后,您一定要以迭代器的形式在一个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);
  }
}

如果你有更多问题,请告诉我。

相关问题