一、初步探索Partitioner

1.1 再次回顧Map階段五大步驟

　　在第四篇博文《初識MapReduce》中，我們認識了MapReduce的八大步湊，其中在Map階段總共五個步驟，如下圖所示：

　　其中，step1.3就是一個分割槽操作。通過前面的學習我們知道Mapper最終處理的鍵值對<key, value>，是需要送到Reducer去合併的，合併的時候，有相同key的鍵/值對會送到同一個Reducer節點中進行歸併。哪個key到哪個Reducer的分配過程，是由Partitioner規定的。在一些叢集應用中，例如分散式快取叢集中，快取的資料大多都是靠雜湊函式來進行資料的均勻分佈的，在Hadoop中也不例外。

1.2 Hadoop內建Partitioner

　　MapReduce的使用者通常會指定Reduce任務和Reduce任務輸出檔案的數量（R）。使用者在中間key上使用分割槽函式來對資料進行分割槽，之後在輸入到後續任務執行程序。一個預設的分割槽函式式使用hash方法（比如常見的：hash(key) mod R）進行分割槽。hash方法能夠產生非常平衡的分割槽，鑑於此，Hadoop中自帶了一個預設的分割槽類HashPartitioner，它繼承了Partitioner類，提供了一個getPartition的方法，它的定義如下所示：

/** Partition keys by their {
 
@link Object#hashCode()}. */
public class HashPartitioner<K, V> extends Partitioner<K, V> {
  /** Use {@link Object#hashCode()} to partition. */
  public int getPartition(K key, V value,
                          int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  }

}
 

　　現在我們來看看HashPartitoner所做的事情，其關鍵程式碼就一句：(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks; 

　　這段程式碼實現的目的是將key均勻分佈在Reduce Tasks上，例如：如果Key為Text的話，Text的hashcode方法跟String的基本一致，都是採用的Horner公式計算，得到一個int整數。但是，如果string太大的話這個int整數值可能會溢位變成負數，所以和整數的上限值Integer.MAX_VALUE（即0111111111111111）進行與運算，然後再對reduce任務個數取餘，這樣就可以讓key均勻分佈在reduce上。 

二、自己定製Partitioner

　　大部分情況下，我們都會使用預設的分割槽函式HashPartitioner。但有時我們又有一些特殊的應用需求，所以我們需要定製Partitioner來完成我們的業務。這裡以第五篇—自定義資料型別處理手機上網日誌為例，來對其中的日誌內容做一個特殊的分割槽：

　　從上圖中我們可以發現，在第二列上並不是所有的資料都是手機號（例如：84138413並不是一個手機號），我們任務就是在統計手機流量時，將手機號碼和非手機號輸出到不同的檔案中。

2.1 自定義KpiPartitioner

    /*
     * 自定義Partitioner類
     */
    public static class KpiPartitioner extends Partitioner<Text, KpiWritable> {
        @Override
        public int getPartition(Text key, KpiWritable value, int numPartitions) {
            // 實現不同的長度不同的號碼分配到不同的reduce task中
            int numLength = key.toString().length();
            if (numLength == 11) {
                return 0;
            } else {
                return 1;
            }
        }
    }

　　這裡按手機和非手機號碼的區分是按該欄位的長度來劃分，如果是11位則為手機號。接下來，就是重新修改run方法中的程式碼：設定為打包執行，設定Partitioner為KpiPartitioner，設定ReducerTask的個數為2；

    public int run(String[] args) throws Exception {
        // 首先刪除輸出目錄已生成的檔案
        FileSystem fs = FileSystem.get(new URI(INPUT_PATH), getConf());
        Path outPath = new Path(OUTPUT_PATH);
        if (fs.exists(outPath)) {
            fs.delete(outPath, true);
        }
        // 定義一個作業
        Job job = new Job(getConf(), "MyKpiJob");
        // 分割槽需要設定為打包執行
        job.setJarByClass(MyKpiJob.class);
        // 設定輸入目錄
        FileInputFormat.setInputPaths(job, new Path(INPUT_PATH));
        // 設定自定義Mapper類
        job.setMapperClass(MyMapper.class);
        // 指定<k2,v2>的型別
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(KpiWritable.class);
        // 設定Partitioner
        job.setPartitionerClass(KpiPartitioner.class);
        job.setNumReduceTasks(2);
        // 設定Combiner
        job.setCombinerClass(MyReducer.class);
        // 設定自定義Reducer類
        job.setReducerClass(MyReducer.class);
        // 指定<k3,v3>的型別
        job.setOutputKeyClass(Text.class);
        job.setOutputKeyClass(KpiWritable.class);
        // 設定輸出目錄
        FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));
        // 提交作業
        System.exit(job.waitForCompletion(true) ? 0 : 1);
        return 0;
    }

注意：分割槽的例子必須要設定為打成jar包執行！

2.2 打成jar包並在Hadoop中執行

　　（1）通過Eclipse匯出jar包

　　（2）通過FTP上傳到Linux中，可以使用各種FTP工具，我一般使用XFtp。

　　（3）通過Hadoop Shell執行jar包中的程式

　　（4）檢視執行結果檔案：

　　首先是part-r-00000，它展示了手機號碼的統計結果

　　然後是part-r-00001，它展示了非手機號碼的統計結果

　　（5）通過Web介面驗證Partitioner的執行：通過訪問http://hadoop-master:50030

　　①是否有2個Reduce任務？

　　從圖中可以看出，總共有2個Reduce任務；

　　②Reduce輸出結果是否一致？

　　手機號碼有20條記錄，一致！

　　非手機號碼只有1條記錄，一致！

總結：分割槽Partitioner主要作用在於以下兩點

（1）根據業務需要，產生多個輸出檔案；

（2）多個reduce任務併發執行，提高整體job的執行效率

參考資料

作者：周旭龍

本文版權歸作者和部落格園共有，歡迎轉載，但未經作者同意必須保留此段宣告，且在文章頁面明顯位置給出原文連結。