hadoop大數據處理之表與表的連接

前言：  hadoop中表連接其實類似於我們用sqlserver對數據進行跨表查詢時運用的inner join一樣，兩個連接的數據要有關系連接起來，中間必須有一個相等的字段進行連接，其實hadoop的表連接就是對文本的處理，處理的文本中有一部分的內容是一樣的，然后把這鞋大量的數據按照中間的一個相同的部分進行連接，用來解決大數據在關系型數據庫查詢困難的問題。

之前一直做c#語言的開發是一個本本分分做網站開發的程序員，像對hadoop這類用java語言做開發的內容一直屬於菜鳥級別，hadoop中表連接也只最近慢慢學的，也是因為要解決工作中的一些問題慢慢熟悉起來的，在工作中確實解決了不少問題，下面我就講一下我在工作中那個地方用到了表連接，其實還有很多地方，這里舉一個比較經典的。

我們公司是做論壇的，數據了談不上太大，不過也還夠我這種小角色忙活半天了，之前出現了一個需求，就是拿到每一個帖子的發帖用戶，和該帖子下所有的回帖用戶，然后基於這個數據統計和發帖用戶最相關的回帖用戶，其實就是找用戶與用戶之間的關系。

這里的數據有兩份（1）發帖人的信息，postid（帖子的id）和userid（發帖人id）（2）回帖人信息， postid（帖子的id）和userid（回帖人id）。這兩個數據在數據庫中是分別放在兩個表中的，看似簡單的問題如果一旦跟大數據扯上了關系就不好處理了，這里的發題人信息由3百萬的數據，而回帖人的信息有7千多萬條數據，中間如果用sqlserver的inner join根本是沒辦法查詢的，這個時候就可以用hadoop的表連接了，首先把數據用工具從數據庫到出成文本，因為要對兩部分數據進行標示，所以兩份數據要在文本中用"\t"分割的之后的length必須不同，

這一份是主帖的數據第一列是postid，第二列是用戶的id，第三列是隨便取出來的數據作為主帖的表示

這一份是回帖的數據，第一列是postid，第二列是回帖的用戶id

在map階段用postid作為key進行，中間給主帖一個標示typeL，並且給回帖的數據進行一個標示typeR。

    private static final Text typeL = new Text("A:");  
    private static final Text typeR = new Text("B:");
    public static class GetMap extends Mapper<Object, Text, Text, MapWritable> {
        public void map(Object key,Text value,Context context) throws IOException,InterruptedException {
            if (value==null) {
                return;
            }
            
            String [] detail=value.toString().split("\t");
            if (detail.length==2) { 
                
                 //回帖信息
                Text threadid=new Text(detail[0]);
                MapWritable mapWritable=new MapWritable();
                mapWritable.put(typeR, new Text(detail[1]));
                context.write(threadid, mapWritable);
            }
            else
            {
                //主帖信息
                Text threadid=new Text(detail[0]);
                MapWritable mapWritable=new MapWritable();
                mapWritable.put(typeL, new Text(detail[1]));
                context.write(threadid, mapWritable);
            }
    
        }
            
        }

在reduce就可以把相關聯的帖子的發帖人id和回帖人id關聯起來了

    public static class GetReduce extends Reducer<Text, MapWritable, Text, Text>{
        public void reduce(Text key,Iterable<MapWritable> values,Context context) throws IOException,InterruptedException{
            Iterator<MapWritable> iterator=values.iterator();
                ArrayList<Text> threadList=new ArrayList<Text>();
                ArrayList<Text> postlist=new ArrayList<Text>();
                while (iterator.hasNext()) {
                    MapWritable mapWritable=iterator.next();
                    if (mapWritable.containsKey(typeL)) {
                        threadList.add((Text) mapWritable.get(typeL));    
                    }else if (mapWritable.containsKey(typeR)) {
                        postlist.add((Text)mapWritable.get(typeR));

                    }
                }
                if (threadList.size()==1&&postlist.size()>0) {
                    String string="";
                    for (Text text : postlist) {
                        string=string+text.toString()+"\t";
                    }
                    context.write(threadList.get(0), new Text(string));
                }
        
        } 
        }

 

最終的結果展示

每一行中，第一個就是發帖人id，后邊跟着的就是回帖人的信息。