Hadoop学习之路（二十一）MapReduce实现Reduce Join（多个文件联合查询）

MapReduce Join

对两份数据data1和data2进行关键词连接是一个很通用的问题，如果数据量比较小，可以在内存中完成连接。

如果数据量比较大，在内存进行连接操会发生OOM。mapreduce join可以用来解决大数据的连接。

1 思路

1.1 reduce join

在map阶段, 把关键字作为key输出，并在value中标记出数据是来自data1还是data2。因为在shuffle阶段已经自然按key分组，reduce阶段，判断每一个value是来自data1还是data2,在内部分成2组，做集合的乘积。

这种方法有2个问题：

1, map阶段没有对数据瘦身，shuffle的网络传输和排序性能很低。

2, reduce端对2个集合做乘积计算，很耗内存，容易导致OOM。

1.2 map join

两份数据中，如果有一份数据比较小，小数据全部加载到内存，按关键字建立索引。大数据文件作为map的输入文件，对map()函数每一对输入，都能够方便地和已加载到内存的小数据进行连接。把连接结果按key输出，经过shuffle阶段，reduce端得到的就是已经按key分组的，并且连接好了的数据。

这种方法，要使用hadoop中的DistributedCache把小数据分布到各个计算节点，每个map节点都要把小数据库加载到内存，按关键字建立索引。

这种方法有明显的局限性：有一份数据比较小，在map端，能够把它加载到内存，并进行join操作。

1.3 使用内存服务器，扩大节点的内存空间

针对map join，可以把一份数据存放到专门的内存服务器，在map()方法中，对每一个<key,value>的输入对，根据key到内存服务器中取出数据，进行连接

1.4 使用BloomFilter过滤空连接的数据

对其中一份数据在内存中建立BloomFilter，另外一份数据在连接之前，用BloomFilter判断它的key是否存在，如果不存在，那这个记录是空连接，可以忽略。

1.5 使用mapreduce专为join设计的包

在mapreduce包里看到有专门为join设计的包，对这些包还没有学习，不知道怎么使用，只是在这里记录下来，作个提醒。

jar： mapreduce-client-core.jar

package： org.apache.hadoop.mapreduce.lib.join

2 实现reduce join

两个文件，此处只写出部分数据，测试数据movies.dat数据量为3883条，ratings.dat数据量为1000210条数据

movies.dat 数据格式为：1　　::　　Toy Story (1995)　　::　　Animation|Children's|Comedy

对应字段中文解释：　　电影ID 　　电影名字　　　　　　　　电影类型

ratings.dat 数据格式为：1　　::　　1193　　::　　5　　::　　978300760

对应字段中文解释：　　用户ID　　电影ID　　　评分　　　　评分时间戳

2个文件进行关联实现代码

import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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 MovieMR1 {

    public static void main(String[] args) throws Exception {
        
        Configuration conf1 = new Configuration();
        /*conf1.set("fs.defaultFS", "hdfs://hadoop1:9000/");
        System.setProperty("HADOOP_USER_NAME", "hadoop");*/
        FileSystem fs1 = FileSystem.get(conf1);
        
        
        Job job = Job.getInstance(conf1);
        
        job.setJarByClass(MovieMR1.class);
        
        job.setMapperClass(MoviesMapper.class);
        job.setReducerClass(MoviesReduceJoinReducer.class);
        
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        Path inputPath1 = new Path("D:\MR\hw\movie\input\movies");
        Path inputPath2 = new Path("D:\MR\hw\movie\input\ratings");
        Path outputPath1 = new Path("D:\MR\hw\movie\output");
        if(fs1.exists(outputPath1)) {
            fs1.delete(outputPath1, true);
        }
        FileInputFormat.addInputPath(job, inputPath1);
        FileInputFormat.addInputPath(job, inputPath2);
        FileOutputFormat.setOutputPath(job, outputPath1);
        
        boolean isDone = job.waitForCompletion(true);
        System.exit(isDone ? 0 : 1);
    }

    
    public static class MoviesMapper extends Mapper<LongWritable, Text, Text, Text>{
        
        Text outKey = new Text();
        Text outValue = new Text();
        StringBuilder sb = new StringBuilder();
        
        protected void map(LongWritable key, Text value,Context context) throws java.io.IOException ,InterruptedException {
            
            FileSplit inputSplit = (FileSplit)context.getInputSplit();
            String name = inputSplit.getPath().getName();
            String[] split = value.toString().split("::");
            sb.setLength(0);
            
            if(name.equals("movies.dat")) {
                //                    1　　::　　Toy Story (1995)　　::　　Animation|Children's|Comedy
                //对应字段中文解释：　　电影ID 　　   电影名字　　　　　　　　                 电影类型
                outKey.set(split[0]);
                StringBuilder append = sb.append(split[1]).append("	").append(split[2]);
                String str = "movies#"+append.toString();
                outValue.set(str);
                //System.out.println(outKey+"---"+outValue);
                context.write(outKey, outValue);
            }else{
                //                    1　　::　　1193　　::　　5　　::　　978300760
                //对应字段中文解释：　　用户ID　　           电影ID　　　      评分　　　　   评分时间戳
                outKey.set(split[1]);
                StringBuilder append = sb.append(split[0]).append("	").append(split[2]).append("	").append(split[3]);
                String str = "ratings#" + append.toString();
                outValue.set(str);
                //System.out.println(outKey+"---"+outValue);
                context.write(outKey, outValue);
            }
        
        };
        
    }
    
    
    public static class MoviesReduceJoinReducer extends Reducer<Text, Text, Text, Text>{
        //用来存放    电影ID    电影名称    电影类型    
        List<String> moviesList = new ArrayList<>();
        //用来存放    电影ID    用户ID 用户评分    时间戳
        List<String> ratingsList = new ArrayList<>();
        Text outValue = new Text();
        
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            
            int count = 0;
            
            //迭代集合
            for(Text text : values) {
                
                //将集合中的元素添加到对应的list中
                if(text.toString().startsWith("movies#")) {
                    String string = text.toString().split("#")[1];
                    
                    moviesList.add(string);
                }else if(text.toString().startsWith("ratings#")){
                    String string = text.toString().split("#")[1];
                    ratingsList.add(string);
                }
            }
            
            //获取2个集合的长度
            long moviesSize = moviesList.size();
            long ratingsSize = ratingsList.size();
            
            for(int i=0;i<moviesSize;i++) {
                for(int j=0;j<ratingsSize;j++) {
                    outValue.set(moviesList.get(i)+"	"+ratingsList.get(j));
                    //最后的输出是    电影ID    电影名称    电影类型    用户ID 用户评分    时间戳
                    context.write(key, outValue);
                }
            }
            
            moviesList.clear();
            ratingsList.clear();
            
        }
        
    }
    
}    

最后的合并结果：　　电影ID　　电影名称　　电影类型　　用户ID　　用户评论　　时间戳

3 实现map join

两个文件，此处只写出部分数据，测试数据movies.dat数据量为3883条，ratings.dat数据量为1000210条数据

movies.dat 数据格式为：1　　::　　Toy Story (1995)　　::　　Animation|Children's|Comedy

对应字段中文解释：　　电影ID 　　电影名字　　　　　　　　电影类型

ratings.dat 数据格式为：1　　::　　1193　　::　　5　　::　　978300760

对应字段中文解释：　　用户ID　　电影ID　　　评分　　　　评分时间戳

需求：求被评分次数最多的10部电影，并给出评分次数（电影名，评分次数）

实现代码

MovieMR1_1.java

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;

public class MovieRating implements WritableComparable<MovieRating>{
    private String movieName;
    private int count;
    
    public String getMovieName() {
        return movieName;
    }
    public void setMovieName(String movieName) {
        this.movieName = movieName;
    }
    public int getCount() {
        return count;
    }
    public void setCount(int count) {
        this.count = count;
    }
    
    public MovieRating() {}
    
    public MovieRating(String movieName, int count) {
        super();
        this.movieName = movieName;
        this.count = count;
    }
    
    
    @Override
    public String toString() {
        return  movieName + "	" + count;
    }
    @Override
    public void readFields(DataInput in) throws IOException {
        movieName = in.readUTF();
        count = in.readInt();
    }
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(movieName);
        out.writeInt(count);
    }
    @Override
    public int compareTo(MovieRating o) {
        return o.count - this.count ;
    }
    
}

MovieMR1_2.java

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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.output.FileOutputFormat;

public class MovieMR1_2 {

    public static void main(String[] args) throws Exception {
        if(args.length < 2) {
            args = new String[2];
            args[0] = "/movie/output/";
            args[1] = "/movie/output_last/";
        }
        
        
        Configuration conf1 = new Configuration();
        conf1.set("fs.defaultFS", "hdfs://hadoop1:9000/");
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        FileSystem fs1 = FileSystem.get(conf1);
        
        
        Job job = Job.getInstance(conf1);
        
        job.setJarByClass(MovieMR1_2.class);
        
        job.setMapperClass(MoviesMapJoinRatingsMapper2.class);
        job.setReducerClass(MovieMR1Reducer2.class);

        
        job.setMapOutputKeyClass(MovieRating.class);
        job.setMapOutputValueClass(NullWritable.class);
        
        job.setOutputKeyClass(MovieRating.class);
        job.setOutputValueClass(NullWritable.class);
        
        
        Path inputPath1 = new Path(args[0]);
        Path outputPath1 = new Path(args[1]);
        if(fs1.exists(outputPath1)) {
            fs1.delete(outputPath1, true);
        }
        //对第一步的输出结果进行降序排序
        FileInputFormat.setInputPaths(job, inputPath1);
        FileOutputFormat.setOutputPath(job, outputPath1);
        
        boolean isDone = job.waitForCompletion(true);
        System.exit(isDone ? 0 : 1);
        

    }
    
    //注意输出类型为自定义对象MovieRating，MovieRating按照降序排序
    public static class MoviesMapJoinRatingsMapper2 extends Mapper<LongWritable, Text, MovieRating, NullWritable>{
        
        MovieRating outKey = new MovieRating();
        
        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            //'Night Mother (1986)         70
            String[] split = value.toString().split("	");
            
            outKey.setCount(Integer.parseInt(split[1]));;
            outKey.setMovieName(split[0]);
            
            context.write(outKey, NullWritable.get());
                        
        }
                
    }
    
    //排序之后自然输出，只取前10部电影
    public static class MovieMR1Reducer2 extends Reducer<MovieRating, NullWritable, MovieRating, NullWritable>{
        
        Text outKey = new Text();
        int count = 0;
        
        @Override
        protected void reduce(MovieRating key, Iterable<NullWritable> values,Context context) throws IOException, InterruptedException {

            for(NullWritable value : values) {
                count++;
                if(count > 10) {
                    return;
                }
                context.write(key, value);
                
            }
        
        }
        
    }
}

MovieRating.java

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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.output.FileOutputFormat;


public class MovieMR1_1 {

    public static void main(String[] args) throws Exception {
        
        if(args.length < 4) {
            args = new String[4];
            args[0] = "/movie/input/";
            args[1] = "/movie/output/";
            args[2] = "/movie/cache/movies.dat";
            args[3] = "/movie/output_last/";
        }
        
        
        Configuration conf1 = new Configuration();
        conf1.set("fs.defaultFS", "hdfs://hadoop1:9000/");
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        FileSystem fs1 = FileSystem.get(conf1);
        
        
        Job job1 = Job.getInstance(conf1);
        
        job1.setJarByClass(MovieMR1_1.class);
        
        job1.setMapperClass(MoviesMapJoinRatingsMapper1.class);
        job1.setReducerClass(MovieMR1Reducer1.class);
        
        job1.setMapOutputKeyClass(Text.class);
        job1.setMapOutputValueClass(IntWritable.class);
        
        job1.setOutputKeyClass(Text.class);
        job1.setOutputValueClass(IntWritable.class);
        
        
        
        //缓存普通文件到task运行节点的工作目录
        URI uri = new URI("hdfs://hadoop1:9000"+args[2]);
        System.out.println(uri);
        job1.addCacheFile(uri);
        
        Path inputPath1 = new Path(args[0]);
        Path outputPath1 = new Path(args[1]);
        if(fs1.exists(outputPath1)) {
            fs1.delete(outputPath1, true);
        }
        FileInputFormat.setInputPaths(job1, inputPath1);
        FileOutputFormat.setOutputPath(job1, outputPath1);
        
        boolean isDone = job1.waitForCompletion(true);
        System.exit(isDone ? 0 : 1);
       
    }
    
    public static class MoviesMapJoinRatingsMapper1 extends Mapper<LongWritable, Text, Text, IntWritable>{
        
        //用了存放加载到内存中的movies.dat数据
        private static Map<String,String> movieMap =  new HashMap<>();
        //key：电影ID
        Text outKey = new Text();
        //value：电影名+电影类型
        IntWritable outValue = new IntWritable();
        
        
        /**
         * movies.dat:    1::Toy Story (1995)::Animation|Children's|Comedy
         * 
         * 
         * 将小表(movies.dat)中的数据预先加载到内存中去
         * */
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            
            Path[] localCacheFiles = context.getLocalCacheFiles();
            
            String strPath = localCacheFiles[0].toUri().toString();
            
            BufferedReader br = new BufferedReader(new FileReader(strPath));
            String readLine;
            while((readLine = br.readLine()) != null) {
                
                String[] split = readLine.split("::");
                String movieId = split[0];
                String movieName = split[1];
                String movieType = split[2];
                
                movieMap.put(movieId, movieName+"	"+movieType);
            }
            
            br.close();
        }
        
        
        /**
         * movies.dat:    1    ::    Toy Story (1995)    ::    Animation|Children's|Comedy    
         *                 电影ID    电影名字                    电影类型
         * 
         * ratings.dat:    1    ::    1193    ::    5    ::    978300760
         *                 用户ID    电影ID        评分        评分时间戳
         * 
         * value:    ratings.dat读取的数据
         * */
        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            
            String[] split = value.toString().split("::");
            
            String userId = split[0];
            String movieId = split[1];
            String movieRate = split[2];
            
            //根据movieId从内存中获取电影名和类型
            String movieNameAndType = movieMap.get(movieId);
            String movieName = movieNameAndType.split("	")[0];
            String movieType = movieNameAndType.split("	")[1];
            
            outKey.set(movieName);
            outValue.set(Integer.parseInt(movieRate));
            
            context.write(outKey, outValue);
            
        }
            
    }

    
    public static class MovieMR1Reducer1 extends Reducer<Text, IntWritable, Text, IntWritable>{
        //每部电影评论的次数
        int count;
        //评分次数
        IntWritable outValue = new IntWritable();
        
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
            
            count = 0;
            
            for(IntWritable value : values) {
                count++;
            }
            
            outValue.set(count);
            
            context.write(key, outValue);
        }
        
    }
    
    
}

最后的结果