``` 對【Hadoop序列化對象 -> 自定義序列化對象 -> 案例：統計手機流量信息】 這個案例添加分區。 ``` 1. 需求：要求將統計結果按照條件輸出到不同文件中（分區）。比如：將統計結果按照手機歸屬地不同省份輸出到不同文件中（分區）。 2. 默認 partition 分區 ```java public class HashPartitioner<K, V> extends Partitioner<K, V> { public int getPartition(K key, V value, int numReduceTasks) { return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks; } } ``` 默認分區是<ins>根據 key 的 hashCode 對 reduceTasks 個數取模得到的</ins>。用戶沒法控制哪個 key 存儲到哪個分區。 3. 自定義 Partitioner 步驟 （1）自定義類繼承 Partitioner，重寫 getPartition()方法。 ```java package com.exa.mapreduce001.flow; import com.exa.mapreduce001.flow.FlowBean; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Partitioner; public class ProvincePartitioner extends Partitioner<Text, FlowBean> { public int getPartition(Text text, FlowBean flowBean, int i) { String preNum = text.toString().substring(0, 3); int partition = 4; if ("136".equals(preNum)) { partition = 0; } if ("137".equals(preNum)) { partition = 1; } if ("138".equals(preNum)) { partition = 2; } if ("139".equals(preNum)) { partition = 3; } return partition; } } ``` （2）在 job 驅動中，設置自定義 Partitioner并指定reduceTask數量。 ```java package com.exa.mapreduce001.flow; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; /** * Date: 2020/12/30 */ public class FlowsumDriver { public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { // 1 獲取配置信息，或者 job 對象實例 Configuration conf = new Configuration(); Job job = Job.getInstance(conf); // 2. 指定本程序的 jar 包所在的本地路徑 job.setJarByClass(FlowsumDriver.class); // 3. 指定本業務 job 要使用的 mapper/Reducer 業務類 job.setMapperClass(FlowCountMapper.class); job.setReducerClass(FlowCountReducer.class); // 4. 指定 mapper 輸出數據的 kv 類型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(FlowBean.class); // 5. 指定最終輸出的數據的 kv 類型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(FlowBean.class); // ===========添加分區和指定reduceTask數量========== job.setPartitionerClass(ProvincePartitioner.class); job.setNumReduceTasks(5); // 6. 指定 job 的輸入/輸出目錄 // 輸入目錄要已經存在 FileInputFormat.setInputPaths(job, new Path("file:///E:\\hadoop\\input")); // 輸出目錄不能已經存在 FileOutputFormat.setOutputPath(job, new Path("file:///E:\\hadoop\\output")); // 7. 將 job 中配置的相關參數，以及 job 所用的 java 類所在的 jar 包，提交給 yarn 去運行 boolean result = job.waitForCompletion(true); System.exit(result ? 0 : 1); } } ``` （3）得出如下5個分區文件 ``` E:\hadoop\output\part-r-00000 E:\hadoop\output\part-r-00001 E:\hadoop\output\part-r-00002 E:\hadoop\output\part-r-00003 E:\hadoop\output\part-r-00004 ``` **注意：** 如果 `reduceTask 的數量>getPartition 的結果數`，則會多產生幾個空的輸出文件 part-r-000xx； 如果 `1<reduceTask 的數量<getPartition` 的結果數，則有一部分分區數據無處安放，會 Exception； 如果 `reduceTask 的數量=1`，則不管 mapTask 端輸出多少個分區文件，最終結果都交給這一個 reduceTask，最終也就只會產生一個結果文件 part-r-00000； 例如：假設自定義分區數為 5，則 （1）job.setNumReduceTasks(1);會正常運行，只不過會產生一個輸出文件 （2）job.setNumReduceTasks(2);會報錯 （3）job.setNumReduceTasks(6);大于 5，程序會正常運行，會產生空文件