在本例中，我们仍然采用上一例中的数据文件。之所以存在reduce side join，是因为在map阶段不能获取所有需要的join字段，即：同一个key对应的字段可能位于不同map中。Reduce side join是非常低效的，因为shuffle阶段要进行大量的数据传输。Map side join是针对以下场景进行的优化：两个待连接表中，有一个表非常大，而另一个表非常小，以至于小表可以直接存放到内存中。这样，我们可以将小表复制多份，让每个map task内存中存在一份（比如存放到hash table中），然后只扫描大表：对于大表中的每一条记录key/value，在hash table中查找是否有相同的key的记录，如果有，则连接后输出即可。为了支持文件的复制，Hadoop提供了一个类DistributedCache，使用该类的方法如下：（1）用户使用静态方法DistributedCache.addCacheFile()指定要复制的文件，它的参数是文件的URI（如果是HDFS上的文件，可以这样：hdfs://jobtracker:50030/home/XXX/file）。JobTracker在作业启动之前会获取这个URI列表，并将相应的文件拷贝到各个TaskTracker的本地磁盘上。（2）用户使用DistributedCache.getLocalCacheFiles()方法获取文件目录，并使用标准的文件读写API读取相应的文件。

本实例中的运行参数需要三个，加入在hdfs中有两个目录input和input2，其中input2存放user.csv，input存放order.csv，则运行命令格式如下：hadoop jar xxx.jar JoinWithDistribute input2/user.csv input output。

具体实例如下，此实例我们采用旧的API来写

public class JoinWithDistribute extends Configured implements Tool
{public static class MapClass extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text>{//用于缓存小表的数据，在这里我们缓存user.csv文件中的数据private Map<String, String> users = new HashMap<String, String>();private Text outKey = new Text();private Text outValue = new Text();//此方法会在map方法执行之前执行@Overridepublic void configure(JobConf job){BufferedReader in = null;try{//从当前作业中获取要缓存的文件Path[] paths = DistributedCache.getLocalCacheFiles(job);String user = null;String[] userInfo = null;for (Path path : paths){if (path.toString().contains("user.csv")){in = new BufferedReader(new FileReader(path.toString()));while (null != (user = in.readLine())){userInfo = user.split(",", 2);//缓存文件中的数据users.put(userInfo[0], userInfo[1]);}}}}catch (IOException e){e.printStackTrace();}finally{try{in.close();}catch (IOException e){e.printStackTrace();}}}public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException{//首先获取order文件中每条记录的userId，//再去缓存中取得相同userId的user记录，合并两记录并输出之。String[] order = value.toString().split(",");String user = users.get(order[0]);if(user != null){outKey.set(user);outValue.set(order[1]);output.collect(outKey, outValue);}}}public int run(String[] args) throws Exception{JobConf job = new JobConf(getConf(), JoinWithDistribute.class);job.setJobName("JoinWithDistribute");job.setMapperClass(MapClass.class);job.setNumReduceTasks(0);job.setInputFormat(TextInputFormat.class);job.setOutputFormat(TextOutputFormat.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(Text.class);//我们把第一个参数的地址作为要缓存的文件路径DistributedCache.addCacheFile(new Path(args[0]).toUri(), job);FileInputFormat.setInputPaths(job, new Path(args[1]));FileOutputFormat.setOutputPath(job, new Path(args[2]));JobClient.runJob(job);return 0;}public static void main(String[] args) throws Exception{int res = ToolRunner.run(new Configuration(), new JoinWithDistribute(), args);System.exit(res);}}

转发:https://blog.csdn.net/huashetianzu/article/details/7821674