hadoop - 表加入MapReduce-Hadoop

Question

我是Hadoop的新手，正在编写我的第一个程序以连接MapReduce中的以下两个表。

第一表:

11111   John
22222   Robert  
33333   Stephan
44444   Peter
55555   Andersen

第二张表:

11111   Washington    EEE   2011
22222   Jacksonville  EIE   2010
33333   Minneapolis   ECE   2012
44444   Cheyenne      CSE   2013
55555   Detroit       IT    2014

我已使用Hue将以上两个文本文件上传到HDFS。每列之间都有一个制表符空间。
运行代码后，我得到了意外的输出，如下所示:

11111   John    Washington  EEE 2011        
22222   Jacksonville    EIE 2010        Robert  
33333   Stephan Minneapolis ECE 2012        
44444   Cheyenne    CSE 2013        Peter   
55555   Andersen    Detroit     IT  2014

我不知道我的代码出了什么问题。这是我的Java代码:

DriverClass:

public class DriverClass extends Configured{

public static void main (String args[]) throws IOException, ClassNotFoundException, InterruptedException{

    Job job = new Job();

    job.setJarByClass(DriverClass.class);


    MultipleInputs.addInputPath(job, new Path(args[0]), TextInputFormat.class, MapperClassOne.class);
    MultipleInputs.addInputPath(job, new Path(args[1]), TextInputFormat.class, MapperClassTwo.class);
    FileOutputFormat.setOutputPath(job, new Path(args[2]));

    job.setReducerClass(ReducerClass.class);

    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(Text.class);

    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);

    System.exit(job.waitForCompletion(true)? 0 : -1);

}
}

我的第一个数据集(第一个表)的MapperClass-MapperClassOne:

public class MapperClassOne extends Mapper<LongWritable, Text, Text, Text>{

public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{

    String[] line = value.toString().split("\t");


    context.write(new Text(line[0]), new Text(line[1]));
}
}

我的第二个数据集(第二个表)的MapperClass-MapperClassTwo:

public class MapperClassTwo extends Mapper<LongWritable, Text, Text, Text>{

public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{

    String[] line = value.toString().split("\t");

    String temp = "";
    for(int i=1; i<line.length; i++){
        temp += line[i] + "\t";
    }

    context.write(new Text(line[0]), new Text(temp));
}
}

ReducerClass:

public class ReducerClass extends Reducer<Text, Text, Text, Text>{

public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException{

    Iterator<Text> iter = values.iterator();

    String temp = "";
    while(iter.hasNext()){
        temp += iter.next().toString() + "\t";
    }

    context.write(key, new Text(temp));

}
}

请帮助我，也建议我是否有更好的方法执行表连接。

Answer 1

在化简器中，除非实现辅助排序，否则不对键的值进行排序。在当前的实现中，键的值可以按任意顺序排列。您需要在映射器值中添加一个标识符，以标识化简器中键的值源。

引用:http://kickstarthadoop.blogspot.com/2011/09/joins-with-plain-map-reduce.html
http://www.lichun.cc/blog/2012/05/hadoop-genericwritable-sample-usage/