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