package duogemap
;
import java.io.IOException
;
import java.util.ArrayList
;
import java.util.List
;
import org.apache.hadoop.conf.Configuration
;
import org.apache.hadoop.fs.Path
;
import org.apache.hadoop.io.LongWritable
;
import org.apache.hadoop.io.Text
;
import org.apache.hadoop.mapreduce.Job
;
import org.apache.hadoop.mapreduce.Mapper
;
import org.apache.hadoop.mapreduce.Reducer
;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
;
import org.apache.hadoop.mapreduce.lib.input.FileSplit
;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat
;
public class JoinJob {
public static final String DELIMITER =
"\u0009";
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
//多路径判断
if (args.length < 2) {
System.out.
println("参数数量不对，至少两个以上参数：<数据文件输出路径>、<输入路径...>");
System.
exit(1
);
}
//输出结果路径
String dataOutput = args[0]
;
//多个路输入径
String[] inputs = new String[args.length - 1]
;
System.
arraycopy(args, 1, inputs, 0, inputs.length
);
Configuration conf = new
Configuration();
Job job = Job.
getInstance(conf,
"join 测试");
job.
setJarByClass(JoinJob.class
);
job.
setMapperClass(MyMapper.class
);
job.
setReducerClass(MyReducer.class
);
job.
setMapOutputKeyClass(Text.class
);
job.
setMapOutputValueClass(Text.class
);
job.
setOutputKeyClass(Text.class
);
job.
setOutputValueClass(Text.class
);
//将输出路径和输入路径放入Path中
Path[] inputPathes = new Path[inputs.length]
;
for
(int i = 0
; i < inputs.length
; i++) {
inputPathes[i] = new
Path(inputs[i]
);
}
Path outputPath = new
Path(dataOutput
);
FileInputFormat.
setInputPaths(job, inputPathes
);
FileOutputFormat.
setOutputPath(job, outputPath
);
job.
waitForCompletion(true
);
}
static class MyMapper extends Mapper<LongWritable, Text, Text, Text> {
private String inputPath
;
private String fileCode =
"";
protected void setup(Context context) throws IOException, InterruptedException {
// 每个文件传进来时获得文件中属性前缀
FileSplit input =
(FileSplit
) context.
getInputSplit();
inputPath = input.
getPath().
getName();
try {
//获得文件名
fileCode = inputPath.
split("_")[0]
;
} catch (Exception e) {
e.
printStackTrace();
}
}
@Override
public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String[] values = value.
toString().
split(DELIMITER
);
StringBuffer sb = new
StringBuffer();
//将文件名拼接到value中，做reduce的判断标识
sb.
append(fileCode +
"#");
boolean first = true
;
for (String v : values) {
if (!first) {
sb.
append(v + DELIMITER
);
}
first = false
;
}
context.
write(new
Text(values[0]
), new
Text(sb.
toString().
substring(0, sb.
toString().
length() - 1
)));
}
}
static class MyReducer extends Reducer<Text, Text, Text, Text> {
@Override
public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
List<String> left = new ArrayList<String>
();
List<String> right = new ArrayList<String>
();
for (Text value : values) {
String[] vv = value.
toString().
split("#");
String fileCode = vv[0]
;
if (fileCode.equals("A.txt")) {
// 左表数据
left.
add(vv[1]
);
} else {
// 右表数据
right.
add(vv[1]
);
}
}
//只有当left和right都有数据是才会遍历
for (String l : left) {
for (String r : right) {
context.
write(new
Text(key
), new
Text(l + DELIMITER + r
));
}
}
}
}
}
//首先准备数据：
//
//假设我们有2张表：
//
//表A（左表）数据：
//
//1 a
//2 b
//3 c
//3 d
//4 e
//6 f
//表B（右表）数据：
//1 10
//2 20
//3 30
//4 40
//4 400
//5 50
//我们需要得到的结果是：
//1 a 10
//2 b 20
//3 c 30
//3 d 30
//4 e 40
//4 e 400
