详解HDFS多文件Join操作的实例

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.contrib.utils.join.DataJoinMapperBase;
import org.apache.hadoop.contrib.utils.join.DataJoinReducerBase;
import org.apache.hadoop.contrib.utils.join.TaggedMapOutput;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


import cn.eshore.traffic.hadoop.util.CommUtil;
import cn.eshore.traffic.hadoop.util.StringUtil;




/**
 * @ClassName: DataJoin
 * @Description: HDFS JOIN操作
 * @author hadoop
 * @date 2012-12-18 下午5:51:32
 */
public class InstallJoin extends Configured implements Tool {
private String static enSplitCode = "\\|";
private String static splitCode = "|";


// 自定义Reducer
public static class ReduceClass extends DataJoinReducerBase {


@Override
protected TaggedMapOutput combine(Object[] tags, Object[] values) {
String joinedStr = "";
//该段判断用户生成Left join限制【其中tags表示文件的路径，install表示文件名称前缀】
//去掉则为All Join
if (tags.length == 1 && tags[0].toString().contains("install")) {
return null;
}

Map map = new HashMap();
for (int i = 0; i