Hadoop架構：MapReduce基本原理和入門案例

本文源碼： GitHub·點這裡 || GitEE·點這裡

一、MapReduce概述

1、基本概念

Hadoop核心元件之一：分布式計算的方案MapReduce，是一種程式設計模型，用于大規模資料集的并行運算，其中Map（映射）和Reduce（歸約）。

MapReduce既是一個程式設計模型，也是一個計算元件，處理的過程分為兩個階段，Map階段：負責把任務分解為多個小任務，Reduce負責把多個小任務的處理結果進行彙總。其中Map階段主要輸入是一對Key-Value，經過map計算後輸出一對Key-Value值；然後将相同Key合并，形成Key-Value集合；再将這個Key-Value集合轉入Reduce階段，經過計算輸出最終Key-Value結果集。

2、特點描述

MapReduce可以實作基于上千台伺服器并發工作，提供很強大的資料處理能力，如果其中單台服務挂掉，計算任務會自動轉義到另外節點執行，保證高容錯性；但是MapReduce不适應于實時計算與流式計算，計算的資料是靜态的。

二、操作案例

1、流程描述

資料檔案一般以CSV格式居多，資料行通常以空格分隔，這裡需要考慮資料内容特點；

檔案經過切片配置設定在不同的MapTask任務中并發執行；

MapTask任務執行完畢之後，執行ReduceTask任務，依賴Map階段的資料；

ReduceTask任務執行完畢後，輸出檔案結果。

2、基礎配置

hadoop:
  # 讀取的檔案源
  inputPath: hdfs://hop01:9000/hopdir/javaNew.txt
  # 該路徑必須是程式運作前不存在的
  outputPath: /wordOut           

3、Mapper程式

public class WordMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    Text mapKey = new Text();
    IntWritable mapValue = new IntWritable(1);

    @Override
    protected void map (LongWritable key, Text value, Context context)
                        throws IOException, InterruptedException {
        // 1、讀取行
        String line = value.toString();
        // 2、行内容切割，根據檔案中分隔符
        String[] words = line.split(" ");
        // 3、存儲
        for (String word : words) {
            mapKey.set(word);
            context.write(mapKey, mapValue);
        }
    }
}           

4、Reducer程式

public class WordReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    int sum ;
    IntWritable value = new IntWritable();

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values,Context context)
                        throws IOException, InterruptedException {
        // 1、累加求和統計
        sum = 0;
        for (IntWritable count : values) {
            sum += count.get();
        }
        // 2、輸出結果
        value.set(sum);
        context.write(key,value);
    }
}           

5、執行程式

@RestController
public class WordWeb {

    @Resource
    private MapReduceConfig mapReduceConfig ;

    @GetMapping("/getWord")
    public String getWord () throws IOException, ClassNotFoundException, InterruptedException {
        // 聲明配置
        Configuration hadoopConfig = new Configuration();
        hadoopConfig.set("fs.hdfs.impl",
                org.apache.hadoop.hdfs.DistributedFileSystem.class.getName()
        );
        hadoopConfig.set("fs.file.impl",
                org.apache.hadoop.fs.LocalFileSystem.class.getName()
        );
        Job job = Job.getInstance(hadoopConfig);

        // Job執行作業 輸入路徑
        FileInputFormat.addInputPath(job, new Path(mapReduceConfig.getInputPath()));
        // Job執行作業 輸出路徑
        FileOutputFormat.setOutputPath(job, new Path(mapReduceConfig.getOutputPath()));

        // 自定義 Mapper和Reducer 兩個階段的任務處理類
        job.setMapperClass(WordMapper.class);
        job.setReducerClass(WordReducer.class);

        // 設定輸出結果的Key和Value的類型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //執行Job直到完成
        job.waitForCompletion(true);
        return "success" ;
    }
}           

6、執行結果檢視

将應用程式打包放到hop01服務上執行；

java -jar map-reduce-case01.jar           

三、案例分析

1、資料類型

Java資料類型與對應的Hadoop資料序列化類型；

2、核心子產品

Mapper子產品：處理輸入的資料，業務邏輯在map()方法中完成，輸出的資料也是KV格式；

Reducer子產品：處理Map程式輸出的KV資料，業務邏輯在reduce()方法中；

Driver子產品：将程式送出到yarn進行排程，送出封裝了運作參數的job對象；

四、序列化操作

1、序列化簡介

序列化：将記憶體中對象轉換為二進制的位元組序列，可以通過輸出流持久化存儲或者網絡傳輸；

反序列化：接收輸入位元組流或者讀取磁盤持久化的資料，加載到記憶體的對象過程；

Hadoop序列化相關接口：Writable實作的序列化機制、Comparable管理Key的排序問題；

2、案例實作

案例描述：讀取檔案，并對檔案相同的行做資料累加計算，輸出計算結果；該案例示範在本地執行，不把Jar包上傳的hadoop伺服器，驅動配置一緻。

實體對象屬性

public class AddEntity implements Writable {

    private long addNum01;
    private long addNum02;
    private long resNum;

    // 構造方法
    public AddEntity() {
        super();
    }
    public AddEntity(long addNum01, long addNum02) {
        super();
        this.addNum01 = addNum01;
        this.addNum02 = addNum02;
        this.resNum = addNum01 + addNum02;
    }

    // 序列化
    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeLong(addNum01);
        dataOutput.writeLong(addNum02);
        dataOutput.writeLong(resNum);
    }
    // 反序列化
    @Override
    public void readFields(DataInput dataInput) throws IOException {
        // 注意：反序列化順序和寫序列化順序一緻
        this.addNum01  = dataInput.readLong();
        this.addNum02 = dataInput.readLong();
        this.resNum = dataInput.readLong();
    }
    // 省略Get和Set方法
}           

Mapper機制

public class AddMapper extends Mapper<LongWritable, Text, Text, AddEntity> {

    Text myKey = new Text();

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

        // 讀取行
        String line = value.toString();

        // 行内容切割
        String[] lineArr = line.split(",");

        // 内容格式處理
        String lineNum = lineArr[0];
        long addNum01 = Long.parseLong(lineArr[1]);
        long addNum02 = Long.parseLong(lineArr[2]);

        myKey.set(lineNum);
        AddEntity myValue = new AddEntity(addNum01,addNum02);

        // 輸出
        context.write(myKey, myValue);
    }
}           

Reducer機制

public class AddReducer extends Reducer<Text, AddEntity, Text, AddEntity> {

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

        long addNum01Sum = 0;
        long addNum02Sum = 0;

        // 處理Key相同
        for (AddEntity addEntity : values) {
            addNum01Sum += addEntity.getAddNum01();
            addNum02Sum += addEntity.getAddNum02();
        }

        // 最終輸出
        AddEntity addRes = new AddEntity(addNum01Sum, addNum02Sum);
        context.write(key, addRes);
    }
}           

案例最終結果：

五、源代碼位址

GitHub·位址
https://github.com/cicadasmile/big-data-parent
GitEE·位址
https://gitee.com/cicadasmile/big-data-parent