数据准备

1. create table if not exists m_ys_lab_jointest_a (
2. id bigint,
3. name string
4. )
5. row format delimited
6. fields terminated by '9'
7. lines terminated by '10'
8. stored as textfile;

1. create table if not exists m_ys_lab_jointest_b (
2. id bigint,
3. statyear bigint,
4. num bigint
5. )
6. row format delimited
7. fields terminated by '9'
8. lines terminated by '10'
9. stored as textfile;

计算模型

1. import java.io.IOException;
2. import java.util.HashMap;
3. import java.util.Iterator;
4. import java.util.Vector;
5. import org.apache.hadoop.io.LongWritable;
6. import org.apache.hadoop.io.Text;
7. import org.apache.hadoop.io.Writable;
8. import org.apache.hadoop.mapred.FileSplit;
9. import org.apache.hadoop.mapred.JobConf;
10. import org.apache.hadoop.mapred.MapReduceBase;
11. import org.apache.hadoop.mapred.Mapper;
12. import org.apache.hadoop.mapred.OutputCollector;
13. import org.apache.hadoop.mapred.RecordWriter;
14. import org.apache.hadoop.mapred.Reducer;
15. import org.apache.hadoop.mapred.Reporter;
16. /**
17. * MapReduce实现Join操作
18. */
19. public class MapRedJoin {
20. public static final String DELIMITER = "\u0009"; // 字段分隔符
21. // map过程
22. public static class MapClass extends MapReduceBase implements
23. Mapper<LongWritable, Text, Text, Text> {
24. public void configure(JobConf job) {
25. super.configure(job);
26. }
27. public void map(LongWritable key, Text value, OutputCollector<Text, Text> output,
28. Reporter reporter) throws IOException, ClassCastException {
29. // 获取输入文件的全路径和名称
30. String filePath = ((FileSplit)reporter.getInputSplit()).getPath().toString();
31. // 获取记录字符串
32. String line = value.toString();
33. // 抛弃空记录
34. if (line == null || line.equals("")) return;
35. // 处理来自表A的记录
36. if (filePath.contains("m_ys_lab_jointest_a")) {
37. String[] values = line.split(DELIMITER); // 按分隔符分割出字段
38. if (values.length < 2) return;
39. String id = values[0]; // id
40. String name = values[1]; // name
41. output.collect(new Text(id), new Text("a#"+name));
42. }
43. // 处理来自表B的记录
44. else if (filePath.contains("m_ys_lab_jointest_b")) {
45. String[] values = line.split(DELIMITER); // 按分隔符分割出字段
46. if (values.length < 3) return;
47. String id = values[0]; // id
48. String statyear = values[1]; // statyear
49. String num = values[2]; //num
50. output.collect(new Text(id), new Text("b#"+statyear+DELIMITER+num));
51. }
52. }
53. }
54. // reduce过程
55. public static class Reduce extends MapReduceBase
56. implements Reducer<Text, Text, Text, Text> {
57. public void reduce(Text key, Iterator<Text> values,
58. OutputCollector<Text, Text> output, Reporter reporter)
59. throws IOException {
60. Vector<String> vecA = new Vector<String>(); // 存放来自表A的值
61. Vector<String> vecB = new Vector<String>(); // 存放来自表B的值
62. while (values.hasNext()) {
63. String value = values.next().toString();
64. if (value.startsWith("a#")) {
65. vecA.add(value.substring(2));
66. } else if (value.startsWith("b#")) {
67. vecB.add(value.substring(2));
68. }
69. }
70. int sizeA = vecA.size();
71. int sizeB = vecB.size();
72. // 遍历两个向量
73. int i, j;
74. for (i = 0; i < sizeA; i ++) {
75. for (j = 0; j < sizeB; j ++) {
76. output.collect(key, new Text(vecA.get(i) + DELIMITER +vecB.get(j)));
77. }
78. }
79. }
80. }
81. protected void configJob(JobConf conf) {
82. conf.setMapOutputKeyClass(Text.class);
83. conf.setMapOutputValueClass(Text.class);
84. conf.setOutputKeyClass(Text.class);
85. conf.setOutputValueClass(Text.class);
86. conf.setOutputFormat(ReportOutFormat.class);
87. }
88. }

技术细节