hadoop join之map side join

在本例中，我们仍然采用上一例中的数据文件。之所以存在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方法执行之前执行
        @Override
        public 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);
    }

}