MapReduce对输入多文件的处理（MultipleInputs类、指定map）

MultipleInputs类指定不同的输入文件路径以及输入文化格式（两个文件，指定哪个map处理哪个文件）
 现有两份数据
 phone
 123,good number
 124,common number
 125,bad number

 user
 zhangsan,123
 lisi,124
 wangwu,125

 现在需要把user和phone按照phone number连接起来。得到下面的结果
 zhangsan,123,good number
 lisi,123,common number
 wangwu,125,bad number

 

分析思路

还是相当于两张表的一对一join操作。join时对value设置个Bean(JavaBean实现writablecomparable接口)，key为外键值

join的优化,详见http://blog.csdn.net/u010366796/article/details/44649933，设置KeyBean(外健和标识flag属性)，进行排序

本例中将通过value进行排序，即在value的JavaBean中通过实习CompareTo()方法，完成排序，使得phone表位于首位

 

1.对value实现JavaBean(实现writablecomparable接口)

1. package test.mr.multiinputs;
3. import java.io.DataInput;
4. import java.io.DataOutput;
5. import java.io.IOException;
7. import org.apache.hadoop.io.WritableComparable;
9. /*
10. * 自定义的JavaBean
11. */
12. public class FlagString implements WritableComparable<FlagString> {
13. private String value;
14. private int flag; // 标记 0:表示phone表 1：表示user表
16. public FlagString() {
17. super();
18. // TODO Auto-generated constructor stub
19. }
21. public FlagString(String value, int flag) {
22. super();
23. this.value = value;
24. this.flag = flag;
25. }
27. public String getValue() {
28. return value;
29. }
31. public void setValue(String value) {
32. this.value = value;
33. }
35. public int getFlag() {
36. return flag;
37. }
39. public void setFlag(int flag) {
40. this.flag = flag;
41. }
43. @Override
44. public void write(DataOutput out) throws IOException {
45. out.writeInt(flag);
46. out.writeUTF(value);
48. }
50. @Override
51. public void readFields(DataInput in) throws IOException {
52. this.flag = in.readInt();
53. this.value = in.readUTF();
54. }
56. @Override
57. public int compareTo(FlagString o) {
58. if (this.flag >= o.getFlag()) {
59. if (this.flag > o.getFlag()) {
60. return 1;
61. }
62. } else {
63. return -1;
64. }
65. return this.value.compareTo(o.getValue());
66. }
68. }

2.多map类，map1(实现对phone表文件操作)

1. package test.mr.multiinputs;
3. import java.io.IOException;
5. import org.apache.hadoop.io.LongWritable;
6. import org.apache.hadoop.io.Text;
7. import org.apache.hadoop.mapreduce.Mapper;
9. public class MultiMap1 extends Mapper<LongWritable, Text, Text, FlagString> {
10. private String delimiter; // 定义分隔符,由job端设置
12. @Override
13. protected void setup(
14. Mapper<LongWritable, Text, Text, FlagString>.Context context)
15. throws IOException, InterruptedException {
16. delimiter = context.getConfiguration().get("delimiter", ",");
17. }
19. @Override
20. protected void map(LongWritable key, Text value,
21. Mapper<LongWritable, Text, Text, FlagString>.Context context)
22. throws IOException, InterruptedException {
23. String line = value.toString().trim();
24. if (line.length() > 0) {
25. String[] str = line.split(delimiter);
26. if (str.length == 2) {
27. context.write(new Text(str[0].trim()),
28. new FlagString(str[1].trim(), 0)); // flag=0,表示phone表
29. }
30. }
31. }
32. }

2.map2(实现对user表文件操作)

1. package test.mr.multiinputs;
3. import java.io.IOException;
5. import org.apache.hadoop.io.LongWritable;
6. import org.apache.hadoop.io.Text;
7. import org.apache.hadoop.mapreduce.Mapper;
9. public class MultiMap2 extends Mapper<LongWritable, Text, Text, FlagString> {
10. private String delimiter; // 设置分隔符
12. @Override
13. protected void setup(
14. Mapper<LongWritable, Text, Text, FlagString>.Context context)
15. throws IOException, InterruptedException {
16. delimiter = context.getConfiguration().get("delimiter", ",");
17. }
19. @Override
20. protected void map(LongWritable key, Text value,
21. Mapper<LongWritable, Text, Text, FlagString>.Context context)
22. throws IOException, InterruptedException {
23. String line = value.toString().trim();
24. if (line.length() > 0) {
25. String[] str = line.split(delimiter);
26. if (str.length == 2) {
27. context.write(new Text(str[1].trim()),
28. new FlagString(str[0].trim(), 1)); // flag=1为user表
29. }
30. }
31. }
32. }

3.reduce类

1. package test.mr.multiinputs;
3. import java.io.IOException;
5. import org.apache.hadoop.io.NullWritable;
6. import org.apache.hadoop.io.Text;
7. import org.apache.hadoop.mapreduce.Reducer;
9. public class MultiRedu extends Reducer<Text, FlagString, NullWritable, Text> {
10. private String delimiter; // 设置分隔符
12. @Override
13. protected void setup(
14. Reducer<Text, FlagString, NullWritable, Text>.Context context)
15. throws IOException, InterruptedException {
16. delimiter = context.getConfiguration().get("delimiter", ",");
17. }
19. @Override
20. protected void reduce(Text key, Iterable<FlagString> values,
21. Reducer<Text, FlagString, NullWritable, Text>.Context context)
22. throws IOException, InterruptedException {
23. // 最后输出的格式为: uservalue,key,phonevalue
24. String phoneValue = "";
25. String userValue = "";
26. int num = 0;
27. for (FlagString value : values) {
28. // 第一个即为phone表
29. if (num == 0) {
30. phoneValue = value.getValue();
31. num++;
32. } else {
33. userValue = value.getValue();
34. context.write(NullWritable.get(),
35. new Text(userValue + key.toString() + phoneValue));
36. }
37. }
38. }
39. }

4.job类(关键!!实现多文件的输入格式等)

1. package test.mr.multiinputs;
3. import org.apache.hadoop.conf.Configuration;
4. import org.apache.hadoop.fs.Path;
5. import org.apache.hadoop.io.NullWritable;
6. import org.apache.hadoop.io.Text;
7. import org.apache.hadoop.mapreduce.Job;
8. import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
9. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
10. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
11. import org.apache.hadoop.util.Tool;
12. import org.apache.hadoop.util.ToolRunner;
14. /*
15. * MultipleInputs类指定不同的输入文件路径以及输入文化格式
16. 现有两份数据
17. phone
18. 123,good number
19. 124,common number
20. 123,bad number
22. user
23. zhangsan,123
24. lisi,124
25. wangwu,125
27. 现在需要把user和phone按照phone number连接起来。得到下面的结果
28. zhangsan,123,good number
29. lisi,123,common number
30. wangwu,125,bad number
31. */
32. public class MultiMapMain extends Configuration implements Tool {
33. private String input1 = null; // 定义的多个输入文件
34. private String input2 = null;
35. private String output = null;
36. private String delimiter = null;
38. @Override
39. public void setConf(Configuration conf) {
41. }
43. @Override
44. public Configuration getConf() {
45. return new Configuration();
46. }
48. @Override
49. public int run(String[] args) throws Exception {
50. setArgs(args);
51. checkParam();// 对参数进行检测
53. Configuration conf = new Configuration();
54. Job job = new Job(conf);
55. job.setJarByClass(MultiMapMain.class);
57. job.setMapOutputKeyClass(Text.class);
58. job.setMapOutputValueClass(FlagString.class);
60. job.setReducerClass(MultiRedu.class);
61. job.setOutputKeyClass(NullWritable.class);
62. job.setOutputValueClass(Text.class);
64. // MultipleInputs类添加文件路径
65. MultipleInputs.addInputPath(job, new Path(input1),
66. TextInputFormat.class, MultiMap1.class);
67. MultipleInputs.addInputPath(job, new Path(input2),
68. TextInputFormat.class, MultiMap2.class);
70. FileOutputFormat.setOutputPath(job, new Path(output));
71. job.waitForCompletion(true);
72. return 0;
73. }
75. private void checkParam() {
76. if (input1 == null || "".equals(input1.trim())) {
77. System.out.println("no input phone-data path");
78. userMaunel();
79. System.exit(-1);
80. }
81. if (input2 == null || "".equals(input2.trim())) {
82. System.out.println("no input user-data path");
83. userMaunel();
84. System.exit(-1);
85. }
86. if (output == null || "".equals(output.trim())) {
87. System.out.println("no output path");
88. userMaunel();
89. System.exit(-1);
90. }
91. if (delimiter == null || "".equals(delimiter.trim())) {
92. System.out.println("no delimiter");
93. userMaunel();
94. System.exit(-1);
95. }
97. }
99. // 用户手册
100. private void userMaunel() {
101. System.err.println("Usage:");
102. System.err.println("-i1 input \t phone data path.");
103. System.err.println("-i2 input \t user data path.");
104. System.err.println("-o output \t output data path.");
105. System.err.println("-delimiter data delimiter \t default comma.");
106. }
108. // 对属性进行赋值
109. // 设置输入的格式:-i1 xxx(输入目录) -i2 xxx(输入目录) -o xxx(输出目录) -delimiter x(分隔符)
110. private void setArgs(String[] args) {
111. for (int i = 0; i < args.length; i++) {
112. if ("-i1".equals(args[i])) {
113. input1 = args[++i]; // 将input1赋值为第一个文件的输入路径
114. } else if ("-i2".equals(args[i])) {
115. input2 = args[++i];
116. } else if ("-o".equals(args[i])) {
117. output = args[++i];
118. } else if ("-delimiter".equals(args[i])) {
119. delimiter = args[++i];
120. }
121. }
122. }
124. public static void main(String[] args) throws Exception {
125. Configuration conf = new Configuration();
126. ToolRunner.run(conf, new MultiMapMain(), args); // 调用run方法
127. }
128. }