Sample data: student.txt
1,yaoshuya,25
2,yaoxiaohua,29
3,yaoyuanyie,15
4,yaoshupei,26
Sample data:score.txt
1,yuwen,100
1,shuxue,99
2,yuwen,99
2,shuxue,88
3,yuwen,99
3,shuxue,56
4,yuwen,33
4,shuxue,99
输出文件内容：
1 [yaoshuya,25,yuwen,100]
1 [yaoshuya,25,shuxue,99]
2 [yaoxiaohua,29,yuwen,99]
2 [yaoxiaohua,29,shuxue,88]
3 [yaoyuanyie,15,yuwen,99]
3 [yaoyuanyie,15,shuxue,56]
4 [yaoshupei,26,yuwen,33]
4 [yaoshupei,26,shuxue,99]
参数：
args= "-Dio.sort.mb=10
-r 1
-inFormat org.apache.hadoop.mapred.KeyValueTextInputFormat
-outFormat org.apache.hadoop.mapred.TextOutputFormat
-outKey org.apache.hadoop.io.Text
-outValue org.apache.hadoop.mapred.join.TupleWritable
hdfs://namenode:9000/user/hadoop/student/student.txt
hdfs://namenode:9000/user/hadoop/student/score2.txt
hdfs://namenode:9000/user/hadoop/joinout".split(" ");
需要注意的是我使用的输出格式是TextOutputFormat（完全是为了方便观察输出后的数据）
输出的valuetype是org.apache.hadoop.mapred.join.TupleWritable ,这个 类型非常方便，类似于数组类型，可以接受多值。
在源码中添加的一句代码，是用来配置我的数据源文件的keyvalue分隔符是,(comma).
jobConf.set("key.value.separator.in.input.line", ",");
关键代码简析：
job.setInputFormatClass(CompositeInputFormat.class);
job.getConfiguration().set(CompositeInputFormat.JOIN_EXPR,
CompositeInputFormat.compose(op, inputFormatClass,
plist.toArray(new Path[0])));
使用CompositeInputFormat来进行join操作。此类的说明：
/**
* An InputFormat capable of performing joins over a set of data sources sorted
* and partitioned the same way.
*
* A user may define new join types by setting the property
* <tt>mapreduce.join.define.<ident></tt> to a classname.
* In the expression <tt>mapreduce.join.expr</tt>, the identifier will be
* assumed to be a ComposableRecordReader.
* <tt>mapreduce.join.keycomparator</tt> can be a classname used to compare
* keys in the join.
* @see #setFormat
* @see JoinRecordReader
* @see MultiFilterRecordReader
*/
通过op来指定连接类型：inner,outer,tbl等，有其他需要也可以实现。
具体是怎么连接的呢？根据两个source进入mapper的key进行归并连接。所以要求数据源是根据key值有序的。此连接是在map端实现的。
测试中我使用KeyValueTextInputFormat来处理，其默认格式是key\tValue,所以我使用了上面的代码来进行重置这个格式。但如果你的文件不是key放在第一个位置，你就需要自己写FileInputFormat啦。
但明显需要你要处理的数据源都是使用同样的FileInputFormat去读取。
还有一点，这里支持多文件连接，示例中我只使用了两个示例文件，可以添加更多的文件，路径添加到outputdir之前即可。