大資料技術之_05_Hadoop學習_02_MapReduce_MapReduce框架原理+InputFormat資料輸入+MapReduce工作...

摘要： 第3章 MapReduce框架原理 3.1 InputFormat資料輸入 3.1.1 切片與MapTask並行度決定機制 1、問題引出 MapTask的並行度決定Map階段的任務處理併發度，進而影響到整個Job的處理速度。 思考： 1G的資料，...

第3章 MapReduce框架原理

3.1 InputFormat資料輸入

3.1.1 切片與MapTask並行度決定機制

1、問題引出

MapTask的並行度決定Map階段的任務處理併發度，進而影響到整個Job的處理速度。

思考： 1G的資料，啟動8個MapTask，可以提高叢集的併發處理能力。那麼1K的資料，也啟動8個MapTask，會提高叢集效能嗎？MapTask並行任務是否越多越好呢？哪些因素影響了MapTask並行度？

2、MapTask並行度決定機制

資料塊：Block是HDFS物理上把資料分成一塊一塊的。

資料切片：資料切片只是在邏輯上對輸入進行分片，並不會在磁碟上將其切分成片進行儲存。

3.1.2 Job提交流程原始碼和切片原始碼詳解

1、Job提交流程原始碼詳解，如下圖所示：

boolean result = job.waitForCompletion(true);

submit();

// 1、建立連線
connect();
// 1）建立提交Job的代理
new Cluster(getConfiguration());
// （1）判斷是本地yarn還是遠端
initialize(jobTrackAddr, conf); 

// 2、提交job
submitter.submitJobInternal(Job.this, cluster)
// 1）建立給叢集提交資料的Stag路徑
Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);

// 2）獲取jobid，並建立job路徑
JobID jobId = submitClient.getNewJobID();

// 3）拷貝jar包到叢集
copyAndConfigureFiles(job, submitJobDir);
rUploader.uploadFiles(job, jobSubmitDir);

// 4）計算切片，生成切片規劃檔案
writeSplits(job, submitJobDir);
maps = writeNewSplits(job, jobSubmitDir);
input.getSplits(job);

// 5）向Stag路徑寫XML配置檔案
writeConf(conf, submitJobFile);
conf.writeXml(out);

// 6）提交job，返回提交狀態
status = submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());

如下圖所示：

2、FileInputFormat切片原始碼解析(input.getSplits(job))

3.1.3 FileInputFormat切片機制

FileInputFormat切片機制

FileInputFormat切片大小的引數配置

3.1.4 CombineTextInputFormat切片機制

框架預設的TextInputFormat切片機制是對任務按檔案規劃切片， 不管檔案多小，都會是一個單獨的切片 ，都會交給一個MapTask，這樣如果有大量小檔案，就 會產生大量的MapTask ，處理效率極其低下。之前處理小檔案(har檔案)使用的方法是歸檔。

• 1、應用場景：
  CombineTextInputFormat用於小檔案過多的場景，它可以將多個小檔案從邏輯上規劃到一個切片中，這樣，多個小檔案就可以交給一個MapTask處理。
• 2、虛擬儲存切片最大值設定
  CombineTextInputFormat.setMaxInputSplitSize(job, 4194304); // 4m
  注意： 虛擬儲存切片最大值設定最好根據實際的小檔案大小情況來設定具體的值。
• 3、切片機制
  生成切片過程包括：虛擬儲存過程和切片過程二部分。
  

（1）虛擬儲存過程：

將輸入目錄下所有檔案大小，依次和設定的setMaxInputSplitSize值比較，如果不大於設定的最大值，邏輯上劃分一個塊。如果輸入檔案大於設定的最大值且大於兩倍，那麼以最大值切割一塊；當剩餘資料大小超過設定的最大值且不大於最大值2倍，此時將檔案均分成2個虛擬儲存塊（防止出現太小切片）。

例如setMaxInputSplitSize值為4M，輸入檔案大小為8.02M，則先邏輯上分成一個4M。剩餘的大小為4.02M，如果按照4M邏輯劃分，就會出現0.02M的小的虛擬儲存檔案，所以將剩餘的4.02M檔案切分成（2.01M和2.01M）兩個檔案。

（2）切片過程：

（a）判斷虛擬儲存的檔案大小是否大於setMaxInputSplitSize值，大於等於則單獨形成一個切片。

（b）如果不大於則跟下一個虛擬儲存檔案進行合併，共同形成一個切片。

（c）測試舉例：有4個小檔案大小分別為1.7M、5.1M、3.4M以及6.8M這四個小檔案，則虛擬儲存之後形成6個檔案塊，大小分別為：

1.7M，（2.55M、2.55M），3.4M以及（3.4M、3.4M）

最終會形成3個切片，大小分別為：

（1.7+2.55）M，（2.55+3.4）M，（3.4+3.4）M

3.1.5 CombineTextInputFormat案例實操

1、需求

將輸入的大量小檔案合併成一個切片統一處理。

（1）輸入資料

準備4個小檔案

（2）期望

期望一個切片處理4個檔案

2、實現過程

（1）不做任何處理，執行1.6節的WordCount案例程式，觀察切片個數為4。

（2）在WordcountDriver中增加如下程式碼，執行程式，並觀察執行的切片個數為3。

（a）驅動類中新增程式碼如下：

// 如果不設定InputFormat，它預設用的是TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class);

// 虛擬儲存切片最大值設定4m
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);

（b）執行如果為3個切片。

（3）在WordcountDriver中增加如下程式碼，執行程式，並觀察執行的切片個數為1。

（a）驅動中新增程式碼如下：

// 如果不設定InputFormat，它預設用的是TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class);

// 虛擬儲存切片最大值設定20m
CombineTextInputFormat.setMaxInputSplitSize(job, 20971520);

（b）執行如果為1個切片。

3.1.6 FileInputFormat實現類

Ctrl + t 可得：

1、TextInputFormat

2、KeyValueTextInputFormat

3、NLineInputFormat

3.1.7 KeyValueTextInputFormat使用案例

1、需求

統計輸入檔案中每一行的第一個單詞相同的行數。

（1）輸入資料

banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang

（2）期望結果資料

banzhang2
xihuan2

2、需求分析

3、程式碼實現

（1）編寫Mapper類

package com.atguigu.mr.kv;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class KVTextMapper extends Mapper<Text, Text, Text, LongWritable> {

// 1、設定value
LongWritable v = new LongWritable(1);

@Override
protected void map(Text key, Text value, Context context)
throws IOException, InterruptedException {
// banzhang ni hao --> banzhang,1
// 2、寫出
context.write(key, v);
}
}

（2）編寫Reducer類

package com.atguigu.mr.kv;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class KVTextReducer extends Reducer<Text, LongWritable, Text, LongWritable> {

LongWritable v = new LongWritable();

@Override
protected void reduce(Text key, Iterable<LongWritable> values, 
Context context) throws IOException, InterruptedException {

// bangzhang,1
// bangzhang,1

Long sum = 0L;
// 1、彙總統計
for (LongWritable value : values) {
sum += value.get();
}
v.set(sum);

// bangzhang,2
// 2、輸出
context.write(key, v);
}
}

（3）編寫Driver類

package com.atguigu.mr.kv;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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.input.KeyValueLineRecordReader;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class KVTextDriver {

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

// 輸入輸出路徑需要根據自己電腦上實際的輸入輸出路徑設定
args = new String[] { "d:/temp/atguigu/0529/input/inputkv", "d:/temp/atguigu/0529/output5" };

Configuration configuration = new Configuration();

// 設定切割符
configuration.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, " ");

// 1、獲取job物件
Job job = Job.getInstance(configuration);

// 2、設定jar包位置
job.setJarByClass(KVTextDriver.class);

// 3、關聯mapper和reducer
job.setMapperClass(KVTextMapper.class);
job.setReducerClass(KVTextReducer.class);

// 4、設定map輸出的kv型別
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(LongWritable.class);

// 5、設定最終輸出資料的kv型別
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(LongWritable.class);

// 6、設定輸入輸出路徑
FileInputFormat.setInputPaths(job, new Path(args[0]));

// 設定輸入格式
job.setInputFormatClass(KeyValueTextInputFormat.class);

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

// 7、提交job
boolean result = job.waitForCompletion(true);
System.exit(result ? 0 : 1);
}
}

3.1.8 NLineInputFormat使用案例

1、需求

對每個單詞進行個數統計，要求根據每個輸入檔案的行數來規定輸出多少個切片。此案例要求每三行放入一個切片中。

（1）輸入資料

banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang banzhang ni hao
xihuan hadoop banzhang

（2）期望輸出資料

Number of splits:4

2、需求分析

3、程式碼實現

（1）編寫Mapper類

package com.atguigu.mr.nline;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class NLineMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

private Text k = new Text();
private LongWritable v = new LongWritable(1);

@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
// 1、獲取一行
String line = value.toString();

// 2、切割
String[] splited = line.split(" ");

// 3、迴圈寫出
for (String string : splited) {
k.set(string);
context.write(k, v);
}
}
}

（2）編寫Reducer類

package com.atguigu.mr.nline;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class NLineReducer extends Reducer<Text, LongWritable, Text, LongWritable> {

LongWritable v = new LongWritable();

@Override
protected void reduce(Text k, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {

Long sum = 0L;
// 1、彙總求和
for (LongWritable value : values) {
sum += value.get();
}

v.set(sum);

// 2、寫出
context.write(k, v);
}
}

（3）編寫Driver類

package com.atguigu.mr.nline;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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.input.NLineInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class NLineDriver {

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
// 輸入輸出路徑需要根據自己電腦上實際的輸入輸出路徑設定
args = new String[] { "d:/temp/atguigu/0529/input/inputnline", "d:/temp/atguigu/0529/output6" };

Configuration configuration = new Configuration();
// 1、獲取job物件
Job job = Job.getInstance(configuration);

// 設定每個切片InputSplit中劃分三條記錄
NLineInputFormat.setNumLinesPerSplit(job, 3);
// 使用NLineInputFormat處理記錄數
job.setInputFormatClass(NLineInputFormat.class);

// 2、設定jar包位置
job.setJarByClass(NLineDriver.class);

// 3、關聯mapper和reducer
job.setMapperClass(NLineMapper.class);
job.setReducerClass(NLineReducer.class);

// 4、設定map輸出的kv型別
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(LongWritable.class);

// 5、設定最終輸出資料的kv型別
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(LongWritable.class);

// 6、設定輸入輸出路徑
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));

// 7、提交job
boolean result = job.waitForCompletion(true);
System.exit(result ? 0 : 1);
}
}

4．測試

（1）輸入資料

banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang
banzhang ni hao
xihuan hadoop banzhang banzhang ni hao
xihuan hadoop banzhang

（2）輸出結果的切片數，如下圖所示：

3.1.9 自定義InputFormat

3.1.10 自定義InputFormat案例實操

無論HDFS還是MapReduce，在 處理小檔案 時效率都非常低，但又難免面臨處理大量小檔案的場景，此時，就需要有相應解決方案。可以自定義InputFormat實現小檔案的合併。

1、需求

將多個小檔案合併成一個SequenceFile檔案（SequenceFile檔案是Hadoop用來儲存二進位制形式的key-value對的檔案格式），SequenceFile裡面儲存著多個檔案，儲存的形式為 檔案路徑+名稱為key ， 檔案內容為value 。

（1）輸入資料

（2）期望輸出檔案格式

2、需求分析

3．程式實現

（1）自定義InputFromat

package com.atguigu.mr.inputformat;

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

public class WholeFileInputformat extends FileInputFormat<Text, BytesWritable> {

@Override
protected boolean isSplitable(JobContext context, Path filename) {
return false;
}

@Override
public RecordReader<Text, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context)
throws IOException, InterruptedException {
WholeRecordReader recordReader = new WholeRecordReader();
recordReader.initialize(split, context);

return recordReader;
}
}

（2）自定義RecordReader類

package com.atguigu.mr.inputformat;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class WholeRecordReader extends RecordReader<Text, BytesWritable> {

private FileSplit split;
private Configuration configuration;
private Text k = new Text();
private BytesWritable v = new BytesWritable();
private boolean isProgress = true;

@Override
public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
// 初始化
this.split = (FileSplit) split;
configuration = context.getConfiguration();
}

@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
// 核心業務邏輯處理
if (isProgress) {
// 1 、定義快取區
byte[] buf = new byte[(int) split.getLength()];

FileSystem fs = null;
FSDataInputStream fis = null;

try {
// 2、獲取檔案系統fs
Path path = split.getPath();
fs = path.getFileSystem(configuration);

// 3、獲取輸入流
fis = fs.open(path);

// 4、讀取檔案內容
IOUtils.readFully(fis, buf, 0, buf.length);

// 5、輸出檔案內容(封裝v)
v.set(buf, 0, buf.length);

// 6、獲取檔案路徑及名稱(封裝k)
k.set(path.toString());

} catch (Exception e) {
} finally {
// 7、關閉資源
IOUtils.closeStream(fis);
}

isProgress = false;

return true;
}

return false;
}

@Override
public Text getCurrentKey() throws IOException, InterruptedException {

return k;
}

@Override
public BytesWritable getCurrentValue() throws IOException, InterruptedException {

return v;
}

@Override
public float getProgress() throws IOException, InterruptedException {
return 0;
}

@Override
public void close() throws IOException {
}
}

（3）編寫SequenceFileMapper類處理流程

package com.atguigu.mr.inputformat;

import java.io.IOException;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class SequenceFileMapper extends Mapper<Text, BytesWritable, Text, BytesWritable> {

@Override
protected void map(Text key, BytesWritable value, Context context)
throws IOException, InterruptedException {

context.write(key, value);
}
}

（4）編寫SequenceFileReducer類處理流程

package com.atguigu.mr.inputformat;

import java.io.IOException;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class SequenceFileReducer extends Reducer<Text, BytesWritable, Text, BytesWritable> {

@Override
protected void reduce(Text key, Iterable<BytesWritable> values,
Context context) throws IOException, InterruptedException {

// 迴圈寫出
for (BytesWritable value : values) {
context.write(key, value);
}
}
}

（5）編寫SequenceFileDriver類處理流程

package com.atguigu.mr.inputformat;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
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 org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;

public class SequenceFileDriver {

public static void main(String[] args) throws IOException, Exception {
// 輸入輸出路徑需要根據自己電腦上實際的輸入輸出路徑設定
args = new String[] { "d:/temp/atguigu/0529/input/inputself", "d:/temp/atguigu/0529/output7" };

// 1、獲取job物件
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);

// 2、設定jar包儲存位置、關聯自定義的mapper和reducer
job.setJarByClass(SequenceFileDriver.class);
job.setMapperClass(SequenceFileMapper.class);
job.setReducerClass(SequenceFileReducer.class);

// 7、設定輸入的inputFormat
job.setInputFormatClass(WholeFileInputformat.class);
// 8、設定輸出的outputFormat
job.setOutputFormatClass(SequenceFileOutputFormat.class);

// 3、設定map輸出端的kv型別
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(BytesWritable.class);

// 4、設定最終輸出端的kv型別
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(BytesWritable.class);

// 5、設定輸入輸出路徑
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));

// 7、提交job
boolean result = job.waitForCompletion(true);

System.exit(result ? 0 : 1);
}
}

3.2 MapReduce工作流程（面試重點）

1、流程示意圖，如下圖所示：

2、流程詳解

上面的流程是整個MapReduce最全工作流程，但是Shuffle過程只是從第7步開始到第16步結束，具體Shuffle過程詳解，如下：

• 1）MapTask收集我們的map()方法輸出的kv對，放到記憶體緩衝區中。
• 2）從記憶體緩衝區不斷溢位本地磁碟檔案，可能會溢位多個檔案。
• 3）多個溢位檔案會被合併成大的溢位檔案。
• 4）在溢位過程及合併的過程中，都要呼叫Partitioner進行分割槽和針對key進行排序。
• 5）ReduceTask根據自己的分割槽號，去各個MapTask機器上取相應的結果分割槽資料。
• 6）ReduceTask會取到同一個分割槽的來自不同MapTask的結果檔案，ReduceTask會將這些檔案再進行合併（歸併排序）。
• 7）合併成大檔案後，Shuffle的過程也就結束了，後面進入ReduceTask的邏輯運算過程（從檔案中取出一個一個的鍵值對Group，呼叫使用者自定義的reduce()方法）。

3、注意

Shuffle中的緩衝區大小會影響到MapReduce程式的執行效率，原則上說，緩衝區越大，磁碟io的次數越少，執行速度就越快。

緩衝區的大小可以通過引數調整，引數：io.sort.mb預設100M。

4、原始碼解析流程

context.write(k, NullWritable.get());
output.write(key, value);
collector.collect(key, value, partitioner.getPartition(key, value, partitions));
HashPartitioner();
collect()
close()
collect.flush()
sortAndSpill()
sort()QuickSort
mergeParts();
file.out
file.out.index
collector.close();

3.3 Shuffle機制（面試重點）

3.3.1 Shuffle機制

Map方法之後，Reduce方法之前的資料處理過程稱之為Shuffle。如下圖所示。

3.3.2 Partition分割槽

自定義Partitioner分割槽步驟：

分割槽總結：

3.3.3 Partition分割槽案例實操

1、需求

將統計結果按照手機歸屬地不同省份輸出到不同檔案中（分割槽）

（1）輸入資料

113736230513 192.196.100.1www.atguigu.com 248124681200
213846544121 192.196.100.22640200
313956435636 192.196.100.31321512200
413966251146 192.168.100.12400404
518271575951 192.168.100.2www.atguigu.com 15272106200
684188413192.168.100.3www.atguigu.com 41161432200
713590439668 192.168.100.41116954200
815910133277 192.168.100.5www.hao123.com31562936200
913729199489 192.168.100.62400200
1013630577991 192.168.100.7www.shouhu.com6960690200
1115043685818 192.168.100.8www.baidu.com36593538200
1215959002129 192.168.100.9www.atguigu.com 1938180500
1313560439638 192.168.100.109184938200
1413470253144 192.168.100.11180180200
1513682846555 192.168.100.12www.qq.com19382910200
1613992314666 192.168.100.13www.gaga.com30083720200
1713509468723 192.168.100.14www.qinghua.com 7335110349404
1818390173782 192.168.100.15www.sogou.com95312412200
1913975057813 192.168.100.16www.baidu.com1105848243200
2013768778790 192.168.100.17120120200
2113568436656 192.168.100.18www.alibaba.com 248124681200
2213568436656 192.168.100.191116954200

（2）期望輸出資料

手機號136、137、138、139開頭都分別放到一個獨立的4個檔案中，其他開頭的放到一個檔案中。

2、需求分析

3、在案例2.4的基礎上，增加一個分割槽類

package com.atguigu.mr.flowsum;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

public class ProvincePartitioner extends Partitioner<Text, FlowBean> {

@Override
public int getPartition(Text key, FlowBean value, int numPartitions) {
// key是手機號，value是流量資訊
// 1、獲取電話號碼的前三位
String prePhoneNum = key.toString().substring(0, 3);
int partition = 4;
// 2、判斷是哪一個省份
if ("136".equals(prePhoneNum)) {
partition = 0;
} else if ("137".equals(prePhoneNum)) {
partition = 1;
} else if ("138".equals(prePhoneNum)) {
partition = 2;
} else if ("139".equals(prePhoneNum)) {
partition = 3;
}

return partition;
}
}

4、在驅動函式中增加自定義資料分割槽設定和ReduceTask設定

package com.atguigu.mr.flowsum;

import java.io.IOException;

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;

public class FlowsumDriver {

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

// 輸入輸出路徑需要根據自己電腦上實際的輸入輸出路徑設定
args = new String[] { "d:/temp/atguigu/0529/input/inputflow", "d:/temp/atguigu/0529/output2" };

// 1、獲取配置資訊，或者獲取job物件例項
Configuration configuration = new Configuration();
Job job = Job.getInstance(configuration);

// 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);

// 8、指定自定義資料分割槽
job.setPartitionerClass(ProvincePartitioner.class);
// 9、同時指定相應數量的reduce task
job.setNumReduceTasks(5);

// 6、指定job的輸入輸出原始檔案所在的目錄
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));

// 7、將job中配置的相關引數，以及job所用的java類所在的jar包，提交給yarn去執行
boolean result = job.waitForCompletion(true);

System.exit(result ? 0 : 1);
}
}