一起學Hadoop——實現兩張表之間的連接操作

---恢復內容開始---

之前我們都是學習使用MapReduce處理一張表的數據（一個文件可視為一張表，hive和關系型數據庫Mysql、Oracle等都是將數據存儲在文件中）。但是我們經常會遇到處理多張表的場景，不同的數據存儲在不同的文件中，因此Hadoop也提供了類似傳統關系型數據庫的join操作。Hadoop生態組件的高級框架Hive、Pig等也都實現了join連接操作，編寫類似SQL的語句，就可以在MapReduce中運行，底層的實現也是基於MapReduce。本文介紹如何使用MapReduce實現join操作，為以后學習hive打下基礎。

1、Map端連。
數據在進入到map函數之前就進行連接操作。適用場景：一個文件比較大，一個文件比較小，小到可以加載到內存中。如果兩個都是大文件，就會出現OOM內存溢出的異常。實現Map端連接操作需要用到Job類的addCacheFile()方法將小文件分發到各個計算節點，然后加載到節點的內存中。

下面通過一個例子來實現Map端join連接操作：
1、雇員employee表數據如下：
name gender age dept_no
Tom male 30 1
Tony male 35 2
Lily female 28 1
Lucy female 32 3

2、部門表dept數據如下：
dept_no dept_name
1 TSD
2 MCD
3 PPD

代碼實現如下:

package join;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.io.*;
import org.apache.hadoop.util.ToolRunner;
import org.apache.hadoop.mapreduce.Mapper;

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.fs.Path;

public class MapJoin extends Configured implements Tool {

    public static class MapJoinMapper extends Mapper<LongWritable, Text, Text,NullWritable> {
        private Map<Integer, String> deptData = new HashMap<Integer, String>();

        @Override
        protected void setup(Mapper<LongWritable, Text, Text,NullWritable>.Context context) throws IOException, InterruptedException {
            super.setup(context);
            //從緩存的中讀取文件。
            Path[] files = context.getLocalCacheFiles();
//            Path file1path = new Path(files[0]);
            BufferedReader reader = new BufferedReader(new FileReader(files[0].toString()));
            String str = null;
            try {
                // 一行一行讀取
                while ((str = reader.readLine()) != null) {
                    // 對緩存中的數據以" "分隔符進行分隔。
                    String[] splits = str.split(" ");
                    // 把需要的數據放在Map中。注意不能操作Map的大小，否則會出現OOM的異常
                    deptData.put(Integer.parseInt(splits[0]), splits[1]);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally{
                reader.close();
            }
        }

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text,NullWritable>.Context context) throws IOException,
                InterruptedException {
            // 獲取從HDFS中加載的表
            String[] values = value.toString().split(" ");
            // 獲取關聯字段depNo，這個字段是關鍵
            int depNo = Integer.parseInt(values[3]);
            // 根據deptNo從內存中的關聯表中獲取要關聯的屬性depName
            String depName = deptData.get(depNo);
            String resultData = value.toString() + " " + depName;
            // 將數據通過context寫入到Reduce中。
            context.write(new Text(resultData),NullWritable.get());
        }
    }

    public static class MapJoinReducer extends Reducer<Text,NullWritable,Text,NullWritable> {
        public void reduce(Text key, Iterable<NullWritable> values,Context context)throws IOException,InterruptedException{
            context.write(key,NullWritable.get());
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Total Sort app");
        //將小表加載到緩存中。
        job.addCacheFile(new URI(args[0]));
        job.setJarByClass(MapJoinMapper.class);
        //1.1 設置輸入目錄和設置輸入數據格式化的類
        FileInputFormat.setInputPaths(job,new Path(args[1]));
        job.setInputFormatClass(TextInputFormat.class);

        //1.2 設置自定義Mapper類和設置map函數輸出數據的key和value的類型
        job.setMapperClass(MapJoinMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        //1.3 設置reduce數量
        job.setNumReduceTasks(1);
        //設置實現了reduce函數的類
        job.setReducerClass(MapJoinReducer.class);

        //設置reduce函數的key值
        job.setOutputKeyClass(Text.class);
        //設置reduce函數的value值
        job.setOutputValueClass(NullWritable.class);

        // 判斷輸出路徑是否存在，如果存在，則刪除
        Path mypath = new Path(args[2]);
        FileSystem hdfs = mypath.getFileSystem(conf);
        if (hdfs.isDirectory(mypath)) {
            hdfs.delete(mypath, true);
        }

        FileOutputFormat.setOutputPath(job, new Path(args[2]));

        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args)throws Exception{

        int exitCode = ToolRunner.run(new MapJoin(), args);
        System.exit(exitCode);
    }
}

執行腳本文件如下：：

/usr/local/src/hadoop-2.6.1/bin/hadoop jar MapJoin.jar \
hdfs://hadoop-master:8020/data/dept.txt \
hdfs://hadoop-master:8020/data/employee.txt \
hdfs://hadoop-master:8020/mapjoin_output

運行結果：

Lily female 28 1 TSD
Lucy female 32 3 PPD
Tom male 30 1 TSD
Tony male 35 2 MCD

 

2、Reduce端連接(Reduce side join)。
數據在Reduce進程中執行連接操作。實現思路：在Map進程中對來自不同表的數據打上標簽，例如來自表employee的數據打上a標簽，來自文件dept表的數據打上b標簽。然后在Reduce進程，對同一個key，來自不同表的數據進行笛卡爾積操作。請看下圖，我們對表employee和表dept的dept_no字段進行關聯，將dept_no字段當做key。

在MapReduce中，key相同的數據會放在一起，因此我們只需在reduce函數中判斷數據是來自哪張表，來自相同表的數據不進行join。

代碼如下：

public class ReduceJoin extends Configured implements Tool {
    public static class JoinMapper extends
            Mapper<LongWritable,Text,Text,Text> {
        String employeeValue = "";
        protected void map(LongWritable key, Text value, Context context)
                throws IOException,InterruptedException {
            /*
             * 根據命令行傳入的文件名，判斷數據來自哪個文件，來自employee的數據打上a標簽，來自dept的數據打上b標簽
             */
            String filepath = ((FileSplit)context.getInputSplit()).getPath().toString();
            String line = value.toString();
            if (line == null || line.equals("")) return;

            if (filepath.indexOf("employee") != -1) {
                String[] lines = line.split(" ");
                if(lines.length < 4) return;

                String deptNo = lines[3];
                employeeValue = line + " a";
                context.write(new Text(deptNo),new Text(employeeValue));
            }

            else if(filepath.indexOf("dept") != -1) {
                String[] lines = line.split(" ");
                if(lines.length < 2) return;
                String deptNo = lines[0];
                context.write(new Text(deptNo), new Text(line + " b"));
            }
        }
    }

    public static class JoinReducer extends
            Reducer<Text, Text, Text, NullWritable> {
        protected void reduce(Text key, Iterable<Text> values,
                              Context context) throws IOException, InterruptedException{
            List<String[]> lista = new ArrayList<String[]>();
            List<String[]> listb = new ArrayList<String[]>();

            for(Text val:values) {
                String[] str = val.toString().split(" ");
                //最后一位是標簽位，因此根據最后一位判斷數據來自哪個文件，標簽為a的數據放在lista中，標簽為b的數據放在listb中
                String flag = str[str.length -1];
                if("a".equals(flag)) {
                    //String valueA = str[0] + " " + str[1] + " " + str[2];
                    lista.add(str);
                } else if("b".equals(flag)) {
                    //String valueB = str[0] + " " + str[1];
                    listb.add(str);
                }
            }

            for (int i = 0; i < lista.size(); i++) {
                if (listb.size() == 0) {
                    continue;
                } else {
                    String[] stra = lista.get(i);
                    for (int j = 0; j < listb.size(); j++) {
                        String[] strb = listb.get(j);
                        String keyValue = stra[0] + " " + stra[1] + " " + stra[2] + " " + stra[3] + " " + strb[1];
                        context.write(new Text(keyValue), NullWritable.get());
                    }
                }
            }
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        GenericOptionsParser optionparser = new GenericOptionsParser(conf, args);
        conf = optionparser.getConfiguration();
        Job job = Job.getInstance(conf, "Reduce side join");
        job.setJarByClass(ReduceJoin.class);
        //1.1 設置輸入目錄和設置輸入數據格式化的類
        //FileInputFormat.setInputPaths(job,new Path(args[0]));
        FileInputFormat.addInputPaths(job, conf.get("input_data"));

        job.setInputFormatClass(TextInputFormat.class);

        //1.2 設置自定義Mapper類和設置map函數輸出數據的key和value的類型
        job.setMapperClass(JoinMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //1.3 設置reduce數量
        job.setNumReduceTasks(1);
        //設置實現了reduce函數的類
        job.setReducerClass(JoinReducer.class);

        //設置reduce函數的key值
        job.setOutputKeyClass(Text.class);
        //設置reduce函數的value值
        job.setOutputValueClass(NullWritable.class);

        // 判斷輸出路徑是否存在，如果存在，則刪除
        Path output_dir = new Path(conf.get("output_dir"));
        FileSystem hdfs = output_dir.getFileSystem(conf);
        if (hdfs.isDirectory(output_dir)) {
            hdfs.delete(output_dir, true);
        }

        FileOutputFormat.setOutputPath(job, output_dir);

        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args)throws Exception{
        int exitCode = ToolRunner.run(new ReduceJoin(), args);
        System.exit(exitCode);
    }
}

執行MapReduce的shell腳本如下：

/usr/local/src/hadoop-2.6.1/bin/hadoop jar ReduceJoin.jar \
-Dinput_data=hdfs://hadoop-master:8020/data/dept.txt,hdfs://hadoop-master:8020/data/employee.txt \
-Doutput_dir=hdfs://hadoop-master:8020/reducejoin_output

總結：
1、Map side join的運行速度比Reduce side join快，因為Reduce side join在shuffle階段會消耗大量的資源。Map side join由於把小表放在內存中，所以執行效率很高。
2、當有一張表的數據很小時，小到可以加載到內存中，那么建議使用Map side join。

 

歡迎關注本人公眾號了解更多關於大數據方面的知識：