大數據技術之_05_Hadoop學習_02_MapReduce_MapReduce框架原理+InputFormat數據輸入+MapReduce工作流程(面試重點)+Shuffle機制(面試重點)

第3章 MapReduce框架原理3.1 InputFormat數據輸入3.1.1 切片與MapTask並行度決定機制3.1.2 Job提交流程源碼和切片源碼詳解3.1.3 FileInputFormat切片機制3.1.4 CombineTextInputFormat切片機制3.1.5 CombineTextInputFormat案例實操3.1.6 FileInputFormat實現類3.1.7 KeyValueTextInputFormat使用案例3.1.8 NLineInputFormat使用案例3.1.9 自定義InputFormat3.1.10 自定義InputFormat案例實操3.2 MapReduce工作流程（面試重點）3.3 Shuffle機制（面試重點）3.3.1 Shuffle機制3.3.2 Partition分區3.3.3 Partition分區案例實操

---

第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）期望結果數據

banzhang    2
xihuan    2

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(); --> Hash分片
        collect()
            close()
                collector.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）輸入數據

1    13736230513 192.196.100.1   www.atguigu.com 2481    24681   200
2    13846544121 192.196.100.2                   264     0       200
3     13956435636 192.196.100.3                   132     1512    200
4     13966251146 192.168.100.1                   240     0       404
5     18271575951 192.168.100.2   www.atguigu.com 1527    2106    200
6     84188413    192.168.100.3   www.atguigu.com 4116    1432    200
7     13590439668 192.168.100.4                   1116    954     200
8     15910133277 192.168.100.5   www.hao123.com  3156    2936    200
9     13729199489 192.168.100.6                   240     0       200
10     13630577991 192.168.100.7   www.shouhu.com  6960    690     200
11     15043685818 192.168.100.8   www.baidu.com   3659    3538    200
12     15959002129 192.168.100.9   www.atguigu.com 1938    180     500
13     13560439638 192.168.100.10                  918     4938    200
14     13470253144 192.168.100.11                  180     180     200
15     13682846555 192.168.100.12  www.qq.com      1938    2910    200
16     13992314666 192.168.100.13  www.gaga.com    3008    3720    200
17     13509468723 192.168.100.14  www.qinghua.com 7335    110349  404
18     18390173782 192.168.100.15  www.sogou.com   9531    2412    200
19     13975057813 192.168.100.16  www.baidu.com   11058   48243   200
20     13768778790 192.168.100.17                  120     120     200
21     13568436656 192.168.100.18  www.alibaba.com 2481    24681   200
22     13568436656 192.168.100.19                  1116    954     200

（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);
    }
}