一、MapReduce概述

1. 定義

MapReduce是一個分布式運算程式的編程框架，使用戶開發“基于Hadoop的資料分析應用”的核心框架，
MapReduce核心功能是將用戶撰寫的業務邏輯代碼和自帶默認組件整合成一個完整的分布式運算程式，并發運行在一個Hadoop集群上，

2. 優勢

<1>易于編程

用戶只關心業務邏輯，實作框架的介面，

<2>良好的擴展性

可以動態增加服務器，解決計算資源不夠的問題，

<3>高容錯性

任何一個節點宕機，可以將任務轉移到其他節點

<4>適合海量資料計算

TB/PB
幾千臺服務器共同計算

3. 劣勢

<1>不適合實時計算

<2>不適合流式計算

sparkstreaming ，flink擅長

<3>不適合DAG有向無環圖計算

spark適合

4. MapReduce核心思想

<1>分為Map階段和Reduce階段
<2>Map階段并發MapTask，完全并行運行，互不干擾
<3>Reduce階段的并發ReduceTask，完全并行運行，但是他們的資料依賴于上一個階段的所有MapTask并發實體的輸出
<4>MapReduce編程模型只能包含一個Map階段和一個Reduce階段，如果用戶的業務邏輯非常復雜，那么只能多個MapReduce程式，串行運行

5. MapReduce行程

<1>MrAppMaster

負責整個程式的程序調度以及狀態協調

<2>MapTask

負責Map階段的整個資料處理流程

<3>ReduceTask

負責Reduce階段的整個資料處理流程

6. WordCount案例

<1>Map類

<2>Reduce類

<3>Main方法

7. 常用資料序列化型別

8. MapReduce編程規范

用戶撰寫的程式分為三部分：Mapper、Reducer 和Driver

<1> Mapper

1. 用戶自定義的Mapper要繼承自己的父類
2. Mapper的輸入資料是kv對的形式
3. Mapper中的業務邏輯寫在map()方法中
4. Mapper的輸出資料是kv對的形式
5. map()方法（MapTask行程）對每一個<k,v>呼叫一次
   

<2> Reducer

1. 用戶自定義的Reducer要繼承自己的父類
2. Reducer的輸入資料型別對應Mapper的輸出資料型別，也是KV對
3. Reducer的業務邏輯寫在reduce()方法中
4. ReduceTask行程對每一組相同k的<k,v>組呼叫一次reduce()方法

<3>Driver

相當于YARN集群的客戶端，用于提交我們整個程式到YRAN集群，提交的是封裝了MapReduce程式相關運行引數的job物件，

1. 獲取配置資訊，獲取job物件實體
2. 指定本程式jar包所在的本地路徑
3. 關聯Mapper/Reducer業務類
4. 指定Mapper輸出資料的kv型別
5. 指定最終輸出的資料的kv型別
6. 指定job的輸入原始檔案所在目錄
7. 指定job的輸出結果所在目錄（不能提前存在）
8. 提交作業

9. WordCount案例實操

1. 需求

<1>在給定的文本檔案中統計輸出每個單詞出現的總次數
<2>輸入資料： hello.txt

<3>期望輸出資料： 統計結果

2. 程序分析

<1>Mapper

1. 將MapTask傳給我們的文本先轉換為String
2. 根據空格將這一行切分成單詞
3. 將單詞輸出為<單詞,1>的形式

<2>Reducer

1. 匯總各個key的個數
2. 輸出該key的總次數

<3>Driver

1. 獲取配置資訊，獲取job物件實體
2. 指定本程式jar包所在的本地路徑
3. 關聯Mapper/Reducer業務類
4. 指定Mapper輸出資料的kv型別
5. 指定最終輸出的資料的kv型別
6. 指定job的輸入原始檔案所在目錄
7. 指定job的輸出結果所在目錄（不能提前存在）
8. 提交作業

3. 環境準備

<1> 新建專案

<2> Maven 坐標

    <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.1.3</version>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.30</version>
        </dependency>
    </dependencies>

<3>Log4J組態檔

log4j.rootLogger=INFO, stdout  
log4j.appender.stdout=org.apache.log4j.ConsoleAppender  
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout  
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n  
log4j.appender.logfile=org.apache.log4j.FileAppender  
log4j.appender.logfile.File=target/spring.log  
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout  
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

<4>創建包路徑

com.demo.mapreduce.wordcount

4. 撰寫程式

<1>WordCountMapper

package com.demo.mapreduce.wordcount;

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

import java.io.IOException;

/**
 * <KEYIN, VALUEIN, KEYOUT, VALUEOUT>
 * KEYIN Map階段輸入的key的型別 ：LongWritable
 * VALUEIN Map階段輸入的value的型別： Text
 * KEYOUT  Map階段輸出的key的型別 ： Text
 * VALUEOUT Map階段輸出的value的型別 : IntWritable
 */
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private Text outK = new Text();
    private IntWritable outV = new IntWritable(1);

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        //1. 獲取一行資料
        String line = value.toString();
        //2. 對資料進行切割
        String[] words = line.split(" ");
        //3. 回圈寫出
        for (String word : words) {
            outK.set(word);
            context.write(outK, outV);
        }
    }
}

<2>WordCountReducer

package com.demo.mapreduce.wordcount;

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

import java.io.IOException;

/**
 * <KEYIN,VALUEIN,KEYOUT,VALUEOUT>
 * KEYIN Reducer階段輸入的key的型別 ：Text
 * VALUEIN Reducer階段輸入的value的型別： IntWritable
 * KEYOUT  Reducer階段輸出的key的型別 ： Text
 * VALUEOUT Reducer階段輸出的value的型別 : IntWritable
 */
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private IntWritable outV = new IntWritable();
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        //1. 累加
        int sum = 0;
        for (IntWritable value : values) {
            sum += value.get();
        }
        //2. 寫出
        outV.set(sum);
        context.write(key, outV);
    }
}

<3>WordCountDriver

package com.demo.mapreduce.wordcount;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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;

/**
 * 1. 獲取job
 * 2. 設定jar包路徑
 * 3. 關聯mapper和reducer
 * 4. 設定map的輸出的key 和 value 型別
 * 5. 設定最終輸出的key 和 value 的型別
 * 6. 指定job的輸入原始檔案所在目錄
 * 7.指定job的輸出結果所在目錄（不能提前存在）
 * 8.提交作業
 */
public class WordCountDriver {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        //1. 獲取job
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        //2. 設定jar包路徑
        job.setJarByClass(WordCountDriver.class);
        //3. 關聯mapper和reducer
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        //4. 設定map的輸出的key 和 value 型別
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        // 5. 設定最終輸出的key 和 value 的型別
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        // 6. 指定job的輸入原始檔案所在目錄
        FileInputFormat.setInputPaths(job, new Path("E:\\javaWordspaces\\MapReduceDemo\hello.txt"));
        //7.指定job的輸出結果所在目錄（不能提前存在）
        FileOutputFormat.setOutputPath(job, new Path("E:\\javaWordspaces\\MapReduceDemo\\result"));
        // 8.提交作業
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }

}

5. 集群運行

<1>maven打包插件

<build>
    <plugins>
        <plugin>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.6.1</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
        <plugin>
            <artifactId>maven-assembly-plugin</artifactId>
            <configuration>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
            </configuration>
            <executions>
                <execution>
                    <id>make-assembly</id>
                    <phase>package</phase>
                    <goals>
                        <goal>single</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

<2>修改WordCountDriver

<3>mvn clean install

<4>將jar包上傳到linux服務器

<5>準備資料

HDFS 下創建 /wcinput/word.txt

<6>執行命令

hadoop jar MapReduceDemo-1.0-SNAPSHOT.jar com.demo.mapreduce.wordcount.WordCountDriver /wcinput/word.txt /wcoutput

<7>結果