MapReduce端Join操作（Map端join、Reduce端join）

柔光的暖阳◎ 2023-02-16 05:28 42阅读 0赞

**map端join**  
指有两种表，只是一张较小，一张较大（一般大于1万条数据），大表的信息完全可以覆盖小表，往往将较小的表以键值对的形式添加到内存中，，然后只扫描大表：对于大表中的每一条记录key/value，在小表中查找是否有相同的key的记录，如果有，则连接后输出即可。  
Map端join是数据到达map处理函数之前进行合并的，效率要远远高于Reduce端join，因为Reduce端join是把所有的数据都经过Shuffle，非常消耗资源。所以一般都用Map端join。  
**操作流程**  
1.以键值对形式存储小表信息；  
2.在Mapper.setup(Context context)函数里读取该文件并存到Map中；  
3.读大表信息，根据外键关联，大表的第二列就是小表的第一列，所以键相同，然后通过大表的第二列的键获取小表中的值，将得到的值将填充回大表中的第二列；  
4.将最后得到的值转为字符串；  
5.将结果输出（即没有Reduce任务）。  
**上代码：**

public class MyMapJoin { 
        public static class MyMapper extends Mapper<LongWritable, Text,Text, NullWritable>{ 
            private Map myType = new HashMap();//存储小表信息
                @Override
                protected void setup(Context context) throws IOException, InterruptedException { 
                  	//获取缓存中的小表，返回的是小表的URI
                    String fileName = context.getCacheFiles()[0].getPath();
               	final BufferedReader read = new BufferedReader(new FileReader(fileName));//buffer读数据增加效率
                	String str = null;
                	while ((str=read.readLine())!=null){ 
                    String [] sps = str.split(",");//通过逗号进行分割
                    myType.put(sps[0],sps[1]);//分割后第一列放到键，第二列放到值
                	}	
                	read.close();
           	    }
           	    @Override
           	    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 
               //读大表
               String [] goodInfos = value.toString().split(",");
               //根据主外键获取大表分类,外键就是第2列，根据外键获取小表第一列的值
               String type = myType.get(goodInfos[1]).toString();
               //将数据填充回大表数组中的第二列
               goodInfos[1] =type;
               //将数组转为字符串
               Text txt = new Text(Arrays.toString(goodInfos));
               context.write(txt,NullWritable.get());
           }
       }
        public static class MyReduce extends Reducer<Text,NullWritable,Text, NullWritable> { 
    
        }
        //主启动类
        public class Driver{ 
    
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException { 
    	Configuration conf = new Configuration();
            //准备一个空任务
            Job job = Job.getInstance(conf);
            //设置该任务的主启动类
            job.setJarByClass(Driver.class);
            //设置任务的输入数据源文件,也就是大表的文件夹
            FileInputFormat.addInputPath(job,new Path("e://xxx"));
            //设置你的mapper任务类
            job.setMapperClass(MyMapper.class);
            //设置mapper任务类的输出数据类型 key和value类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(NullWritable.class);
            //放到linux系统下读hdfs文件的路径
            //job.addCacheFile(new URI("hdfs://192.168.xx.xxx:9000/mydemo/xxx.csv"));
    
            //读本地文件,也就是小表
            job.addFileToClassPath(new Path("e://xxx.csv"));
            //设置任务的输出数据源文件
            FileOutputFormat.setOutputPath(job,new Path("e://gg"));
            //启动任务并执行
            job.waitForCompletion(true);
    
        }
    }
    }

**reduce端join**  
两张都是大表，所以同一个key对应的字段可能位于不同map中，把两张表放在同一目录下，map函数同时读取两个文件File1和File2，为了区分两种来源的key/value数据对，对每条数据打一个标签，File1就用 type: File2就用context: 根据文件名进行判断File1 和File2,由于有外键关联，输出的时候把相同的外键作为键输出。  
在reduce中按分类编号已经分好组 ，处理两种不同的数据先将迭代器中的数据存放到集合中；在数组中找到前面有type:的信息，通过截取得到值；将其它的信息遍历替换中间标签的内容，最后输出。  
**操作流程**  
1.两张表放在同一目录下，map函数同时读取两个文件；  
2.((FileSplit)context.getInputSplit()).getPath().toString()获取当前行数据出自哪个文件的文件名；  
3.path.contains(“type”)判断是哪个文件，把相同的外键作为键输出；  
4.增强型for遍历迭代器中的数据放到集合中；  
5.通过.substring截取有type:的后面的值；  
6.将其它的信息遍历，替换context:的内容变为刚才截取有type:的后面的值；  
7.在主启动类输出。  
**代码**

public class MyReduceInner { 
        public static class MyMapper extends Mapper<LongWritable, Text,Text,Text>{ 
    
            @Override
            protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
                String [] words = value.toString().split(",");
                //获取当前行数据出自哪个文件的文件名
                String path = ((FileSplit)context.getInputSplit()).getPath().toString();
                //不同文件输出的键的位置不同
                if(path.contains("type")){ //两个文件有外键关联，找到外键作为相同的键，放到一组
                    //.contains返回一个值，该值指示指定的 String 对象是否出现在此字符串中。
                    context.write(new Text(words[0]),new Text("type:"+words[1]));//找到文件关联的键的列是第一列
                }else { 
                    context.write(new Text(words[1]),new Text("context:"+words[0]+":"+words[1]+":"+words[2]));//找到文件关联的外键的列的位置是第二个
                }
            }
        }
    
        public static class MyReduce extends Reducer< Text,Text,Text, NullWritable>{ 
            @Override
            protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { 
                //按分类编号已经分好组 处理两种不同的数据
                //先将迭代器中的数据存放到集合中
                //把迭代器转换为集合
                List<Text> lst = new ArrayList<Text>();
                for (Text tx : values) { 
                    String s = tx.toString();//先转为字符串
                    lst.add(new Text(s));
                }
                //在数组中找到前面有type:的信息 用来获得值的信息
                String typeInfo = "";
                for (Text tx : lst) { 
                    String val = tx.toString();
                    if(val.contains("type")){ 
                        typeInfo = val.substring(val.indexOf(":")+1);//截取冒号后面一位
                        //在集合中移除本条信息
                        lst.remove(tx);
                        break;
                    }
                }
                //将其它的信息遍历替换context:的内容变为刚才截取有type:的后面的值
                for (Text tx : lst) { 
                    String [] infos = tx.toString().split(":");
                    infos[2] = typeInfo;//拿type:截取过后的值赋给数组第二个下标，因为外键相同
                    infos[0] = String.valueOf(key);//把外键替换掉标识context:
                    context.write(new Text(Arrays.toString(infos)),NullWritable.get());
                }
            }
        }
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
           
            Configuration conf = new Configuration();
            //准备一个空任务
            Job job = Job.getInstance(conf);
            //设置该任务的主启动类，本类
            job.setJarByClass(MyReduceInner.class);
            //设置任务的输入数据源文件,存2张表的目录
            FileInputFormat.addInputPath(job,new Path("e://xxx"));
    
            //设置你的mapper任务类
            job.setMapperClass(MyMapper.class);
            //设置mapper任务类的输出数据类型 key和value类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);
    
            //设置你的reduce任务类
            job.setReducerClass(MyReduce.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(NullWritable.class);
            //设置任务的输出数据源文件
            FileOutputFormat.setOutputPath(job,new Path("e://xx"));
            //启动任务并执行
            job.waitForCompletion(true);
    
        }
    }