一、Partitioner概述

Map階段總共五個步驟，2就是一個分區操作

?

哪個key到哪個Reducer的分配過程，是由Partitioner規定的。

?

二、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;}}

將key均勻布在Reduce Tasks上
(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;?

如果Key為Text的話，Text的hashcode方法跟String的基本一致，都是采用的Horner公式計算，得到一個int整數。但是，如果string太大的話這個int整數值可能會溢出變成負數，所以和整數的上限值Integer.MAX_VALUE（即0111111111111111）進行與運算，然后再對reduce任務個數取余，這樣就可以讓key均勻分布在reduce上?

?

三、自定制Partitioner

一般我們都會使用默認的分區函數HashPartitioner

自定義數據類型處理手機上網日志： 在第二列上并不是所有的數據都是手機號（84138413并不是一個手機號），任務就是在統計手機流量時，將手機號碼和非手機號輸出到不同的文件中

自定義MKPartitioner

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

設置為打包運行，設置Partitioner為MKPartitioner設置ReducerTask的個數為2
注意：分區的例子必須要設置為打成jar包運行！

? ? public int run(String[] args) throws Exception {// 定義一個作業Job job = new Job(getConf(), "MyJob");// 分區需要設置為打包運行job.setJarByClass(MyJob.class);// 設置輸入目錄FileInputFormat.setInputPaths(job, new Path(INPUT_PATH));// 設置自定義Mapper類job.setMapperClass(MyMapper.class);// 指定<k2,v2>的類型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(KpiWritable.class);// 設置Partitionerjob.setPartitionerClass(NKPartitioner.class);job.setNumReduceTasks(2);// 設置自定義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包并在Hadoop中運行

通過Idea導出jar包

通過FTP上傳到Linux中，可以使用各種FTP工具

通過Hadoop Shell執行jar包中的程序

通過Web接口驗證Partitioner的運行：

• 通過訪問http://hadoop01:50030 查看 是否有2個Reduce任務？
• Reduce輸出結果是否一致？

小結：

• 分區Partitioner主要作用在于以下兩點 ?根據業務需要，產生多個輸出文件
• 多個reduce任務并發運行，提高整體job的運行效率

總結

以上是生活随笔為你收集整理的Hadoop入门（八）Mapreduce高级shuffle之Partitioner的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。