当前位置:操作系统 > Unix/Linux >>

hadoop mapreduce多表关联

hadoop mapreduce多表关联
 
假设有如下两个文件,一个是表是公司和地址的序号的对应,一个表是地址的序号和地址的名称的对应。
 
表1:
 
[plain] 
A:Beijing Red Star  1  
A:Shenzhen Thunder  3  
A:Guangzhou Honda   2  
A:Beijing Rising    1  
A:Guangzhou Development Bank    2  
A:Tencent   3  
A:Back of Beijing   1  
 
表2:
[plain] 
B:1 Beijing  
B:2 Guangzhou  
B:3 Shenzhen  
B:4 Xian  
 
mapreduce如下:
[plain] 
private static final Text typeA = new Text("A:");  
      
    private static final Text typeB = new Text("B:");  
      
    private static Log log = LogFactory.getLog(MTJoin.class);  
      
    public static class Map extends Mapper<Object, Text, Text, MapWritable> {  
          
        public void map(Object key, Text value, Context context)  
                throws IOException, InterruptedException {  
            String valueStr = value.toString();  
            String type = valueStr.substring(0, 2);  
            String content = valueStr.substring(2);  
            log.info(content);  
            if(type.equals("A:"))  
            {  
                String[] contentArray = content.split("\t");  
                String city = contentArray[0];  
                String address = contentArray[1];  
                MapWritable map = new MapWritable();  
                map.put(typeA, new Text(city));  
                context.write(new Text(address), map);  
            }  
            else if(type.equals("B:"))  
            {  
                String[] contentArray = content.split("\t");  
                String adrNum = contentArray[0];  
                String adrName = contentArray[1];  
                MapWritable map = new MapWritable();  
                map.put(typeB, new Text(adrName));  
                context.write(new Text(adrNum), map);  
            }  
        }  
    }  
      
    public static class Reduce extends Reducer<Text, MapWritable, Text, Text> {  
          
          
          
         public void reduce(Text key, Iterable<MapWritable> values, Context context)  
                 throws IOException, InterruptedException {  
             Iterator<MapWritable> it = values.iterator();  
             List<Text> cityList = new ArrayList<Text>();  
             List<Text> adrList = new ArrayList<Text>();  
             while(it.hasNext())  
             {  
                 MapWritable map = it.next();  
                 if(map.containsKey(typeA))  
                 {  
                     cityList.add((Text)map.get(typeA));  
                 }  
                 else if(map.containsKey(typeB))  
                 {  
                     adrList.add((Text)map.get(typeB));  
                 }  
             }  
             for(int i = 0; i < cityList.size(); i++)  
             {  
                 for(int j = 0; j < adrList.size(); j++)  
                 {  
                     context.write(cityList.get(i), adrList.get(j));  
                 }  
             }  
         }  
    }  
原理很简单,map的出口,以地址的序号作为key,然后出来的时候,公司名称放一个list,地址的名称放一个list,两个list的内容作笛卡儿积,就得到了结果。
输出如下:
 
[plain] 
Beijing Red Star    Beijing  
Beijing Rising  Beijing  
Back of Beijing Beijing  
Guangzhou Honda Guangzhou  
Guangzhou Development Bank  Guangzhou  
Shenzhen Thunder    Shenzhen  
Tencent Shenzhen  
 
CopyRight © 2022 站长资源库 编程知识问答 zzzyk.com All Rights Reserved
部分文章来自网络,