一、Partitioner概述

Map階段總共五個(gè)步驟，2就是一個(gè)分區(qū)操作

?

哪個(gè)key到哪個(gè)Reducer的分配過(guò)程，是由Partitioner規(guī)定的。

?

二、Hadoop內(nèi)置Partitioner

MapReduce的使用者通常會(huì)指定Reduce任務(wù)和Reduce任務(wù)輸出文件的數(shù)量（R）。
用戶在中間key上使用分區(qū)函數(shù)來(lái)對(duì)數(shù)據(jù)進(jìn)行分區(qū)，之后在輸入到后續(xù)任務(wù)執(zhí)行進(jìn)程。一個(gè)默認(rèn)的分區(qū)函數(shù)式使用hash方法（比如常見(jiàn)的：hash(key) mod R）進(jìn)行分區(qū)。hash方法能夠產(chǎn)生非常平衡的分區(qū)。

Hadoop中自帶了一個(gè)默認(rèn)的分區(qū)類HashPartitioner，
它繼承了Partitioner類，提供了一個(gè)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為T(mén)ext的話，Text的hashcode方法跟String的基本一致，都是采用的Horner公式計(jì)算，得到一個(gè)int整數(shù)。但是，如果string太大的話這個(gè)int整數(shù)值可能會(huì)溢出變成負(fù)數(shù)，所以和整數(shù)的上限值Integer.MAX_VALUE（即0111111111111111）進(jìn)行與運(yùn)算，然后再對(duì)reduce任務(wù)個(gè)數(shù)取余，這樣就可以讓key均勻分布在reduce上?

?

三、自定制Partitioner

一般我們都會(huì)使用默認(rèn)的分區(qū)函數(shù)HashPartitioner

自定義數(shù)據(jù)類型處理手機(jī)上網(wǎng)日志： 在第二列上并不是所有的數(shù)據(jù)都是手機(jī)號(hào)（84138413并不是一個(gè)手機(jī)號(hào)），任務(wù)就是在統(tǒng)計(jì)手機(jī)流量時(shí)，將手機(jī)號(hào)碼和非手機(jī)號(hào)輸出到不同的文件中

自定義MKPartitioner

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

設(shè)置為打包運(yùn)行，設(shè)置Partitioner為MKPartitioner設(shè)置ReducerTask的個(gè)數(shù)為2
注意：分區(qū)的例子必須要設(shè)置為打成jar包運(yùn)行！

? ? public int run(String[] args) throws Exception {// 定義一個(gè)作業(yè)Job job = new Job(getConf(), "MyJob");// 分區(qū)需要設(shè)置為打包運(yùn)行job.setJarByClass(MyJob.class);// 設(shè)置輸入目錄FileInputFormat.setInputPaths(job, new Path(INPUT_PATH));// 設(shè)置自定義Mapper類job.setMapperClass(MyMapper.class);// 指定<k2,v2>的類型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(KpiWritable.class);// 設(shè)置Partitionerjob.setPartitionerClass(NKPartitioner.class);job.setNumReduceTasks(2);// 設(shè)置自定義Reducer類job.setReducerClass(MyReducer.class);// 指定<k3,v3>的類型job.setOutputKeyClass(Text.class);job.setOutputKeyClass(KpiWritable.class);// 設(shè)置輸出目錄FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));// 提交作業(yè)System.exit(job.waitForCompletion(true) ? 0 : 1);return 0;}

打成jar包并在Hadoop中運(yùn)行

通過(guò)Idea導(dǎo)出jar包

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

通過(guò)Hadoop Shell執(zhí)行jar包中的程序

通過(guò)Web接口驗(yàn)證Partitioner的運(yùn)行：

• 通過(guò)訪問(wèn)http://hadoop01:50030 查看 是否有2個(gè)Reduce任務(wù)？
• Reduce輸出結(jié)果是否一致？

小結(jié)：

• 分區(qū)Partitioner主要作用在于以下兩點(diǎn) ?根據(jù)業(yè)務(wù)需要，產(chǎn)生多個(gè)輸出文件
• 多個(gè)reduce任務(wù)并發(fā)運(yùn)行，提高整體job的運(yùn)行效率

總結(jié)

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

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。