Hadoop 之 MapReduce

1 MapReduce 概述

MapReduce 是一個分布式運算程序的編程框架败京，是用戶開發(fā)基于 Hadoop 的數(shù)據(jù)分析應(yīng)用的核心框架耍鬓。

MapReduce 核心功能是將用戶編寫的業(yè)務(wù)邏輯代碼和自帶默認(rèn)組件整合成一個完整的分布式運算程序，并發(fā)運行在一個Hadoop 集群上马靠。

1.1 MapReduce 優(yōu)缺點

優(yōu)點：

MapReduce 易于編程

它簡單的實現(xiàn)一些接口吼鳞，就可以完成一個分布式程序看蚜，這個分布式程序可以分布到大量廉價的 PC 機(jī)器上運行，也就是說寫一個分布式程序赔桌，跟寫一個簡單的串行程序是一模一樣的供炎，就是因為這個特點使得 MapReduce 編程變得非常流行。
良好的擴(kuò)展性

當(dāng)計算資源不能得到滿足的時候疾党，可以通過簡單的增加機(jī)器來擴(kuò)展它的計算能力音诫。
高容錯性

MapReduce 設(shè)計的初衷就是使程序能夠部署在廉價的 PC 機(jī)器上，這就要求它具有很高的容錯性雪位，比如其中一臺機(jī)器掛了竭钝，它可以把上面的計算任務(wù)轉(zhuǎn)移到另外一個節(jié)點上運行，不至于這個任務(wù)運行失敗，而且這個過程不需要人工參與香罐，而完全是由 Hadoop 內(nèi)部完成的卧波。
適合 PB 級以上海量數(shù)據(jù)的離線處理

可以實現(xiàn)上千臺服務(wù)器集群并發(fā)工作，提供數(shù)據(jù)處理能力庇茫。

缺點：

不擅長實時計算

MapReduce 無法像 MySQL 一樣港粱，在毫秒或者秒級內(nèi)返回結(jié)果。
不擅長流式計算

流式計算的輸入數(shù)據(jù)是動態(tài)的旦签，而 MapReduce 的輸入數(shù)據(jù)集是靜態(tài)的查坪，不能動態(tài)變化。這是因為 MapReduce 自身的設(shè)計特點決定了數(shù)據(jù)源必須是靜態(tài)的顷霹。
不擅長DAG（有向圖）計算

多個應(yīng)用程序存在依賴關(guān)系咪惠，后一個應(yīng)用程序的輸入為前一個的輸出，在這種情況下淋淀，MapReduce 并不是不能做遥昧，而是使用后，每個 MapReduce 作業(yè)的輸出結(jié)果都會寫入到磁盤朵纷，會造成大量的磁盤 IO炭臭，導(dǎo)致性能非常的低下。

1.2 MapReduce 核心思想

image

分布式的運算程序往往需要分成至少 2 個階段袍辞。

第一個階段的 MapTask 并發(fā)實例鞋仍，完全并行運行，互不相干搅吁。

第二個階段的 ReduceTask 并發(fā)實例互不相干威创，但是他們的數(shù)據(jù)依賴于上一個階段的所有 MapTask 并發(fā)實例的輸出。

MapReduce 編程模型只能包含一個 Map 階段和一個 Reduce 階段谎懦，如果用戶的業(yè)務(wù)邏輯非常復(fù)雜肚豺，那就只能多個MapReduce 程序，串行運行界拦。

1.3 MapReduce 進(jìn)程

一個完整的 MapReduce 程序在分布式運行時有三類實例進(jìn)程：

MrAppMaster 負(fù)責(zé)整個程序的過程調(diào)度及狀態(tài)協(xié)調(diào)

MapTask 負(fù)責(zé) Map 階段的整個數(shù)據(jù)處理流程吸申。

ReduceTask 負(fù)責(zé) Reduce 階段的整個數(shù)據(jù)處理流程。

1.4 常用數(shù)據(jù)序列化類型

Java 類型	Hadoop Writable 類型
Boolean	BooleanWritable
Byte	ByteWritable
Int	IntWritable
Float	FloatWritable
Long	LongWritable
Double	DoubleWritable
String	Text
Map	MapWritable
Array	ArrayWritable

1.5 MapReduce 編程規(guī)范

用戶編寫的程序分成三個部分：

Mapper 階段

用戶自定義的 Mapper 要繼承自己的父類
Mapper 的輸入數(shù)據(jù)是 KV 對的形式（KV的類型可自定義
Mapper 中的業(yè)務(wù)邏輯寫在 map() 方法中
Mapper的輸出數(shù)據(jù)是 KV對的形式（KV的類型可自定義）
map() 方法（MapTask 進(jìn)程）對每一個 <k,v> 調(diào)用一次

Reduce 階段

用戶自定義的 Reducer 要繼承自己的父類
Reducer 的輸入數(shù)據(jù)類型對應(yīng) Mapper 的輸出數(shù)據(jù)類型享甸，也是 KV
Reducer 的業(yè)務(wù)邏輯寫在 reduce() 方法中
ReduceTask 進(jìn)程對每一組相同 k 的 <k,v> 組調(diào)用一次reduce()方法

Driver 階段

相當(dāng)于 YARN 集群的客戶端截碴，用于提交我們整個程序到 YARN 集群，提交的是封裝了 MapReduce 程序相關(guān)運行參數(shù)的job對象

1.6 WordCount 案例實操

導(dǎo)入依賴

<dependencies>
    <dependency>
        <groupid>junit</groupid>
        <artifactid>junit</artifactid>
        <version>RELEASE</version>
    </dependency>
    <dependency>
        <groupid>org.apache.logging.log4j</groupid>
        <artifactid>log4j-core</artifactid>
        <version>2.8.2</version>
    </dependency>
    <dependency>
        <groupid>org.apache.hadoop</groupid>
        <artifactid>hadoop-common</artifactid>
        <version>2.7.2</version>
    </dependency>
    <dependency>
        <groupid>org.apache.hadoop</groupid>
        <artifactid>hadoop-client</artifactid>
        <version>2.7.2</version>
    </dependency>
    <dependency>
        <groupid>org.apache.hadoop</groupid>
        <artifactid>hadoop-hdfs</artifactid>
        <version>2.7.2</version>
    </dependency>
    <dependency>
        <groupid>jdk.tools</groupid>
        <artifactid>jdk.tools</artifactid>
        <version>1.8</version>
        <scope>system</scope>
        <systempath>${JAVA_HOME}/lib/tools.jar</systempath>
    </dependency>
</dependencies>

log4j.properties

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

WcMapper

package com.djm.mapreduce;

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;

public class WcMapper extends Mapper<longwritable, text,text, intwritable> {

    private Text key = new Text();

    private IntWritable one = new IntWritable(1);

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] words = line.split(" ");
        for (String word : words) {
            this.key.set(word);
            context.write(this.key, this.one);
        }
    }
}

WcReduce

package com.djm.mapreduce;

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

import java.io.IOException;

public class WcReduce extends Reducer<text, intwritable, text, intwritable> {

    private IntWritable total = new IntWritable();

    @Override
    protected void reduce(Text key, Iterable<intwritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable count : values) {
            sum += 1;
        }
        this.total.set(sum);
        context.write(key, this.total);
    }
}

WcDriver

package com.djm.mapreduce;

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;

public class WcDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // 獲得任務(wù)
        Job job = Job.getInstance(new Configuration());
        // 設(shè)置Classpath
        job.setJarByClass(WcDriver.class);
        // 設(shè)置Mapper
        job.setMapperClass(WcMapper.class);
        // 設(shè)置Reducer
        job.setReducerClass(WcReduce.class);
        // 設(shè)置Mapper的輸出key和value的類型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        // 設(shè)置Reducer的輸出key和value的類型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // 設(shè)置輸入和輸出路徑
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

2 Hadoop 序列化

2.1 為什么不使用 Java 序列化框架進(jìn)行序列化

Serializable 是一個重量級的 Java 序列框架蛉威，一個對象被序列化后日丹，會產(chǎn)生很多額外的信息（各種校驗信息，Header蚯嫌，繼承體系等）聚凹，會產(chǎn)生大量的 IO割坠，所以不適合在網(wǎng)絡(luò)中高效的傳輸，所以湘今，Hadoop 自己開發(fā)了一個輕量級的序列化框架（Writable）。

Hadoop序列化特點：

1摩瞎、緊湊：高效使用存儲空間孝常。

2旗们、快速：讀寫數(shù)據(jù)的額外開銷小

3上渴、可擴(kuò)展：隨著通信協(xié)議的升級而可升級稠氮。

4隔披、互操作：支持多語言的交互。

2.2 自定義 bean 對象實現(xiàn)序列化接口

在開發(fā)過程中往往提供的基本序列化類型不能滿足要求鬓长，一般情況都需要創(chuàng)建一個 Bean 實現(xiàn) Writable 接口痢士。

具體實現(xiàn) bean 對象序列化步驟如下 7 步：

1怠蹂、實現(xiàn) Writable 接口

2城侧、反序列化時豆茫，需要反射調(diào)用空參構(gòu)造函數(shù)揩魂，必須提供空參構(gòu)造

3火脉、重寫序列化方法

4倦挂、重寫反序列方法

5方援、反序列化和序列化的順序必須完全一致

6、要想把結(jié)果顯示在文件中笛丙，需要重寫 toString()

7胚鸯、如果需要將自定義的 bean 放在 key 中傳輸，則還需要實現(xiàn) Comparable 接口哥桥，因為 MapReduce 框中的 Shuffle 過程要求對 key 必須能排序

2.3 序列化案例實操

統(tǒng)計每一個手機(jī)號耗費的總上行流量拟糕、下行流量送滞、總流量

輸入數(shù)據(jù)格式：id 手機(jī)號碼網(wǎng)絡(luò)ip 上行流量下行流量網(wǎng)絡(luò)狀態(tài)碼

輸出數(shù)據(jù)格式：手機(jī)號碼上行流量下行流量總流量

FlowBean

package com.djm.mapreduce.flow;

import org.apache.hadoop.io.Writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class FlowBean implements Writable {

    private long upFlow;

    private long downFlow;

    private long sumFlow;

    public FlowBean() {
    }

    public void set(long upFlow, long downFlow) {
        this.upFlow = upFlow;
        this.downFlow = downFlow;
        this.sumFlow = this.upFlow + this.downFlow;
    }

    public long getUpFlow() {
        return upFlow;
    }

    public void setUpFlow(long upFlow) {
        this.upFlow = upFlow;
    }

    public long getDownFlow() {
        return downFlow;
    }

    public void setDownFlow(long downFlow) {
        this.downFlow = downFlow;
    }

    public long getSumFlow() {
        return sumFlow;
    }

    public void setSumFlow(long sumFlow) {
        this.sumFlow = sumFlow;
    }

    @Override
    public String toString() {
        return upFlow + "\t" + downFlow + "\t" + sumFlow;
    }

    public void write(DataOutput out) throws IOException {
        out.writeLong(upFlow);
        out.writeLong(downFlow);
        out.writeLong(sumFlow);
    }

    public void readFields(DataInput in) throws IOException {
        this.upFlow = in.readLong();
        this.downFlow = in.readLong();
        this.sumFlow = in.readLong();
    }
}

FlowMapper

package com.djm.mapreduce.flow;

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

import java.io.IOException;

public class FlowMapper extends Mapper<longwritable, text, flowbean> {

    private FlowBean flowBean = new FlowBean();
    private Text phone = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String[] words = line.split("\t");
        phone.set(words[1]);
        long upFlow = Long.parseLong(words[words.length - 3]);
        long downFlow = Long.parseLong(words[words.length - 2]);
        flowBean.set(upFlow, downFlow);
        context.write(phone, flowBean);
    }
}

FlowReduce

package com.djm.mapreduce.flow;

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

import java.io.IOException;

public class FlowReduce extends Reducer<text, flowbean, text, flowbean> {

    private FlowBean totalFlow = new FlowBean();
    @Override
    protected void reduce(Text key, Iterable<flowbean> values, Context context) throws IOException, InterruptedException {
        long sumUpFlow = 0;
        long sumDownFlow = 0;
        for (FlowBean value : values) {
            long upFlow = value.getUpFlow();
            long downFlow = value.getDownFlow();
            sumUpFlow += upFlow;
            sumDownFlow += downFlow;
        }
        totalFlow.set(sumUpFlow, sumDownFlow);
        context.write(key, totalFlow);

    }
}

FlowDriver

package com.djm.mapreduce.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;

public class FlowDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Job job = Job.getInstance(new Configuration());
        job.setJarByClass(FlowDriver.class);
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReduce.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

3 MapReduce 框架原理

3.1 InputFormat 數(shù)據(jù)輸入

3.1.1 切片與MapTask并行度決定機(jī)制

一個 Job 的 Map 階段并行度由客戶端在提交 Job 時的切片數(shù)決定

每一個Split切片分配一個MapTask并行實例處理

默認(rèn)情況下褂微，切片大小=BlockSize

切片時不考慮數(shù)據(jù)集整體式撼，而是逐個針對每一個文件單獨切片

3.1.2 FileInputFormat 切片機(jī)制

切片機(jī)制：

簡單的按照文件的內(nèi)容長度進(jìn)行切片
切片大小等于 Block 大小
切片時不考慮數(shù)據(jù)集整體端衰，而是逐個針對每個文件單獨切片

源碼中如何計算切片大小的？

Math.max(minSize, Math.min(maxSize, blockSize));
mapreduce.input.fileinputformat.split.minsize=1 默認(rèn)值為1
mapreduce.input.fileinputformat.split.maxsize= Long.MAXValue 默認(rèn)值Long.MAXValue

如何自定義切片大小忘嫉？

maxsize（切片最大值）：參數(shù)如果調(diào)得比blockSize小庆冕，則會讓切片變小，而且就等于配置的這個參數(shù)的值拷姿。
minsize（切片最小值）：參數(shù)調(diào)的比blockSize大响巢，則可以讓切片變得比blockSize還大。

3.1.3 CombineTextInputFormat 切片機(jī)制

CombineTextInputFormat 用于小文件過多的場景伏穆，它可以將多個小文件從邏輯上規(guī)劃到一個切片中蜈出，這樣偷厦，多個小文件就可以交給一個 MapTask 處理只泼。

image

3.1.4 FileInputFormat 的其他實現(xiàn)類

TextInputFormat：

TextInputForma 是默認(rèn)的 FileInputFormat 實現(xiàn)類，按行讀取每條記錄十绑，鍵是存儲該行在整個文件中的起始字節(jié)偏移量，LongWritable 類型甚亭，值是這行的內(nèi)容偶摔，不包括任何行終止符（換行符和回車符）信不，Text類型。

KeyValueTextInputFormat：

每一行均為一條記錄下硕，被分隔符分割為 key，value誉尖，可以通過在驅(qū)動類中設(shè)置conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, "\t"); 來設(shè)定分隔符琢感，默認(rèn)分隔符是 tab驹针。

NLineInputFormat：

如果使用 NlineInputFormat柬甥，代表每個 map 進(jìn)程處理的 InputSplit 不再按 Block 塊去劃分，而是按 NlineInputFormat 指定的行數(shù)N來劃分撤防，即輸入文件的總行數(shù) /N = 切片數(shù)辜膝，如果不整除厂抖，切片數(shù) = 商 + 1忱辅。

3.1.5 自定義 InputFormat

無論 HDFS 還是 MapReduce橡卤，在處理小文件時效率都非常低碧库，但又難免面臨處理大量小文件的場景，此時沽瞭，就需要有相應(yīng)解決方案城丧≤狡叮可以自定義 InputFormat 實現(xiàn)小文件的合并。

image

程序?qū)崿F(xiàn)：

WholeFileInputformat

package com.djm.mapreduce.inputformat;

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;

import java.io.IOException;

public class WholeFileInputformat extends FileInputFormat<text, byteswritable> {
    @Override
    protected boolean isSplitable(JobContext context, Path filename) {
        return false;
    }

    public RecordReader<text, byteswritable> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        return new WholeRecordReader();
    }
}

WholeRecordReader

package com.djm.mapreduce.inputformat;

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.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;

import java.io.IOException;

public class WholeRecordReader extends RecordReader<text, byteswritable> {

    private boolean notRead  = true;

    private Text key = new Text();

    private BytesWritable value = new BytesWritable();

    private FSDataInputStream fis;

    private FileSplit fs;

    /**
     * 初始化方法俱诸，框架會在開始的時候調(diào)用一次
     * @param split
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        // 轉(zhuǎn)換切換類型為文件切片
        fs = (FileSplit) split;
        // 通過切片獲取文件路徑
        Path path = fs.getPath();
        // 通過路徑獲取文件系統(tǒng)
        FileSystem fileSystem = path.getFileSystem(context.getConfiguration());
        // 開流
        fis = fileSystem.open(path);
    }

    /**
     * 讀取下一組KV
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    public boolean nextKeyValue() throws IOException, InterruptedException {
        if (notRead) {
            // 讀K
            key.set(fs.getPath().toString());
            // 讀V
            byte[] buf = new byte[(int) fs.getLength()];
            fis.read(buf);
            value.set(buf, 0, buf.length);
            notRead = false;
            return true;
        } else {
            return false;
        }
    }

    /**
     * 獲取當(dāng)前讀到的key
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    public Text getCurrentKey() throws IOException, InterruptedException {
        return this.key;
    }

    /**
     * 獲取當(dāng)前讀到的value
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    public BytesWritable getCurrentValue() throws IOException, InterruptedException {
        return this.value;
    }

    /**
     * 當(dāng)前數(shù)據(jù)讀取的進(jìn)度
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    public float getProgress() throws IOException, InterruptedException {
        return notRead ? 0 : 1;
    }

    /**
     * 關(guān)閉資源
     * @throws IOException
     */
    public void close() throws IOException {
        if (fis != null) {
            fis.close();
        }

    }
}

WholeFileDriver

package com.djm.mapreduce.inputformat;

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;

import java.io.IOException;

public class WholeFileDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Job job = Job.getInstance(new Configuration());

        job.setJarByClass(WholeFileDriver.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(BytesWritable.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(BytesWritable.class);

        job.setInputFormatClass(WholeFileInputformat.class);
        job.setOutputFormatClass(SequenceFileOutputFormat.class);

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

        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
    }
}

3.2 MapReduce工作流程

image

上面的流程是整個 MapReduce 最全工作流程园骆，但是 Shuffle 過程只是從第 7 步開始到第 16 步結(jié)束锄码，具體 Shuffle 過程詳解滋捶，如下：

1）MapTask 收集我們的 map() 方法輸出的 KV 對，放到內(nèi)存緩沖區(qū)中

2）從內(nèi)存緩沖區(qū)不斷溢出本地磁盤文件亲族，可能會溢出多個文件

3）多個溢出文件會被合并成大的溢出文件

4）在溢出過程及合并的過程中，都要調(diào)用 Partitioner 進(jìn)行分區(qū)和針對 key 進(jìn)行排序

5）ReduceTask 根據(jù)自己的分區(qū)號知给，去各個 MapTask 機(jī)器上取相應(yīng)的結(jié)果分區(qū)數(shù)據(jù)

6）ReduceTask 會取到同一個分區(qū)的來自不同 MapTask 的結(jié)果文件戈次，ReduceTask 會將這些文件再進(jìn)行合并（歸并排序）

7）合并成大文件后，Shuffle 的過程也就結(jié)束了悬秉，后面進(jìn)入 ReduceTask 的邏輯運算過程（從文件中取出一個一個的鍵值對 Group和泌，調(diào)用用戶自定義的 reduce() 方法）

3.3 Shuffle 機(jī)制

image

3.3.1 Partition 分區(qū)

分區(qū)可以將統(tǒng)計結(jié)果按照條件輸出到不同的文件中

默認(rèn) Partition 分區(qū)：

public class HashPartitioner<k, v> extends Partitioner<k, v> {

    public int getPartition(K key, V value, int numReduceTasks) {
        return (key.hashCode() &amp; Integer.MAX_VALUE) % numReduceTasks;
    }
}

默認(rèn)分區(qū)是根據(jù) key 的 hashCode 對 ReduceTasks 個數(shù)取模決定的。

自定義 Partition 步驟：

自定義類繼承 Partitioner县恕，重寫 getPartition() 方法

public class CustomPartitioner extends Partitioner<text, flowbean> {
    @Override
    public int getPartition(Text key, FlowBean value, int numPartitions) {
        // 控制分區(qū)代碼邏輯
        return partition;
    }
}

在驅(qū)動類中，指定 Partitioner
自定義 Partition 后，要根據(jù)自定義 Partitioner 的邏輯設(shè)置相應(yīng)數(shù)量的 ReduceTask

注意：

如果 ReduceTask 的數(shù)量 > getPartition 的結(jié)果數(shù)，則會多產(chǎn)生幾個空的輸出文件 part-r-000xx奔垦；
如果 1< ReduceTask的數(shù)量 < getPartition 的結(jié)果數(shù)，則有一部分分區(qū)數(shù)據(jù)無處安放犯眠，會 Exception；
如果 ReduceTask 的數(shù)量 = 1量蕊，則不管 MapTask 端輸出多少個分區(qū)文件韭赘，最終結(jié)果都交給這一個 ReduceTask，最終也就只會產(chǎn)生一個結(jié)果文件 part-r-00000瓦灶；
分區(qū)號必須從零開始贼陶，逐一累加。

需求分析：

image

代碼實現(xiàn)：

# ProvincePartitioner
package com.djm.mapreduce.partitioner;

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

public class ProvincePartitioner extends Partitioner<flowbean, text> {

    @Override
    public int getPartition(FlowBean flowBean, Text text, int numPartitions) {
        switch (text.toString().substring(0, 3)) {
            case "136":
                return 0;
            case "137":
                return 1;
            case "138":
                return 2;
            case "139":
                return 3;
            default:
                return 4;
        }
    }
}

# PartitionerFlowDriver
package com.djm.mapreduce.partitioner;

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;

public class PartitionerFlowDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Job job = Job.getInstance(new Configuration());
        job.setJarByClass(PartitionerFlowDriver.class);
        job.setMapperClass(SortMapper.class);
        job.setReducerClass(SortReduce.class);
        job.setMapOutputKeyClass(FlowBean.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);
        job.setPartitionerClass(ProvincePartitioner.class);
        job.setNumReduceTasks(5);
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

3.3.2 WritableComparable 排序

排序是 MapReduce 框架中最重要的操作之一老翘，MapTask 和 ReduceTask 均會對數(shù)據(jù)按照 key 進(jìn)行排序墓怀，該操作屬于Hadoop 的默認(rèn)行為，任何應(yīng)用程序中的數(shù)據(jù)均會被排序钓账，而不管邏輯上是否需要。

默認(rèn)排序是按照字典順序排序惕蹄，且實現(xiàn)該排序的方法是快速排序：

對于 MapTask遭顶，它會將處理的結(jié)果暫時放到環(huán)形緩沖區(qū)中，當(dāng)環(huán)形緩沖區(qū)使用率達(dá)到一定閾值后铣揉，再對緩沖區(qū)中的數(shù)據(jù)進(jìn)行一次快速排序，并將這些有序數(shù)據(jù)溢寫到磁盤上朽合，而當(dāng)數(shù)據(jù)處理完畢后，它會對磁盤上所有文件進(jìn)行歸并排序讲婚。

對于 ReduceTask，它從每個 MapTask 上遠(yuǎn)程拷貝相應(yīng)的數(shù)據(jù)文件，如果文件大小超過一定閾值，則溢寫磁盤上倔丈，否則存儲在內(nèi)存中，如果磁盤上文件數(shù)目達(dá)到一定閾值，則進(jìn)行一次歸并排序以生成一個更大文件蜜氨，如果內(nèi)存中文件大小或者數(shù)目超過一定閾值埋哟，則進(jìn)行一次合并后將數(shù)據(jù)溢寫到磁盤上，當(dāng)所有數(shù)據(jù)拷貝完畢后抛计，ReduceTask 統(tǒng)一對內(nèi)存和磁盤上的所有數(shù)據(jù)進(jìn)行一次歸并排序。

排序分類：

image

需求分析：

image

代碼實現(xiàn)：

package com.djm.mapreduce.partitioner;

import org.apache.hadoop.io.WritableComparable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

@Data
public class FlowBean implements WritableComparable<flowbean> {

    private long upFlow;

    private long downFlow;

    private long sumFlow;

    public void set(long upFlow, long downFlow) {
        this.upFlow = upFlow;
        this.downFlow = downFlow;
        this.sumFlow = this.upFlow + this.downFlow;
    }

    public void write(DataOutput out) throws IOException {
        out.writeLong(upFlow);
        out.writeLong(downFlow);
        out.writeLong(sumFlow);
    }

    public void readFields(DataInput in) throws IOException {
        this.upFlow = in.readLong();
        this.downFlow = in.readLong();
        this.sumFlow = in.readLong();
    }

    @Override
    public int compareTo(FlowBean o) {
        return this.sumFlow &gt; o.sumFlow ? -1:1;
    }
}

3.3.3 GroupingComparator 分組

對 Reduce 階段的數(shù)據(jù)根據(jù)某一個或幾個字段進(jìn)行分組媒抠。

分組排序步驟：

自定義類繼承WritableComparator
重寫compare()方法

創(chuàng)建一個構(gòu)造將比較對象的類傳給父類

protected OrderGroupingComparator() {
    super(OrderBean.class, true);
}

需求分析：

image

代碼實現(xiàn)：

# OrderBean
package com.djm.mapreduce.order;

import org.apache.hadoop.io.WritableComparable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

@Data
public class OrderBean implements WritableComparable<orderbean> {

    private String orderId;

    private String productId;

    private double price;

    @Override
    public int compareTo(OrderBean o) {
        int compare = this.orderId.compareTo(o.orderId);

        if (compare == 0) {
            return Double.compare(o.price, this.price);
        } else {
            return compare;
        }
    }

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(orderId);
        out.writeUTF(productId);
        out.writeDouble(price);
    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.orderId = in.readUTF();
        this.productId = in.readUTF();
        this.price = in.readDouble();
    }
}

# OrderSortGroupingComparator
package com.djm.mapreduce.order;

import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;

public class OrderSortGroupingComparator extends WritableComparator {

    public OrderSortGroupingComparator() {
        super(OrderBean.class, true);
    }

    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        OrderBean oa = (OrderBean) a;
        OrderBean ob = (OrderBean) b;

        return oa.getOrderId().compareTo(ob.getOrderId());
    }
}

# OrderSortDriver
package com.djm.mapreduce.order;

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

public class OrderSortDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Job job = Job.getInstance(new Configuration());
        job.setJarByClass(OrderSortDriver.class);
        job.setMapperClass(OrderSortMapper.class);
        job.setReducerClass(OrderSortReduce.class);
        job.setMapOutputKeyClass(OrderBean.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setGroupingComparatorClass(OrderSortGroupingComparator.class);
        job.setOutputKeyClass(OrderBean.class);
        job.setOutputValueClass(NullWritable.class);
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

3.4 MapTask 工作機(jī)制

image

1）Read 階段：MapTask 通過用戶編寫的 RecordReader刘急，從輸入 InputSplit 中解析出一個個 key/value。

2）Map階段：該節(jié)點主要是將解析出的 key/value 交給用戶編寫 map() 函數(shù)處理，并產(chǎn)生一系列新的 key/value另假。

3）Collect 收集階段：在用戶編寫 map() 函數(shù)中边篮，當(dāng)數(shù)據(jù)處理完成后己莺，一般會調(diào)用 OutputCollector.collect() 輸出結(jié)果，在該函數(shù)內(nèi)部戈轿，它會將生成的 key/value 分區(qū)（調(diào)用Partitioner）凌受，并寫入一個環(huán)形內(nèi)存緩沖區(qū)中。

4）Spill 階段：即溢寫胁艰，當(dāng)環(huán)形緩沖區(qū)滿后，MapReduce 會將數(shù)據(jù)寫到本地磁盤上智蝠，生成一個臨時文件腾么，需要注意的是，將數(shù)據(jù)寫入本地磁盤之前杈湾，先要對數(shù)據(jù)進(jìn)行一次本地排序解虱，并在必要時對數(shù)據(jù)進(jìn)行合并、壓縮等操作漆撞。

利用快速排序算法對緩存區(qū)內(nèi)的數(shù)據(jù)進(jìn)行排序殴泰，排序方式是，先按照分區(qū)編號 Partition 進(jìn)行排序浮驳，然后按照 key 進(jìn)行排序悍汛，這樣，經(jīng)過排序后至会，數(shù)據(jù)以分區(qū)為單位聚集在一起离咐，且同一分區(qū)內(nèi)所有數(shù)據(jù)按照 key 有序。
按照分區(qū)編號由小到大依次將每個分區(qū)中的數(shù)據(jù)寫入任務(wù)工作目錄下的臨時文件 output/spillN.out（N表示當(dāng)前溢寫次數(shù)）中奉件，如果用戶設(shè)置了 Combiner宵蛀，則寫入文件之前，對每個分區(qū)中的數(shù)據(jù)進(jìn)行一次聚集操作县貌。
將分區(qū)數(shù)據(jù)的元信息寫到內(nèi)存索引數(shù)據(jù)結(jié)構(gòu) SpillRecord 中术陶，其中每個分區(qū)的元信息包括在臨時文件中的偏移量、壓縮前數(shù)據(jù)大小和壓縮后數(shù)據(jù)大小煤痕，如果當(dāng)前內(nèi)存索引大小超過 1MB梧宫，則將內(nèi)存索引寫到文件 output/spillN.out.index中。

5）Combine 階段：當(dāng)所有數(shù)據(jù)處理完成后摆碉，MapTask 對所有臨時文件進(jìn)行一次合并塘匣，以確保最終只會生成一個數(shù)據(jù)文件。

6）當(dāng)所有數(shù)據(jù)處理完后兆解，MapTask 會將所有臨時文件合并成一個大文件，并保存到文件 output/file.out 中跑揉，同時生成相應(yīng)的索引文件 output/file.out.index锅睛。

7）在進(jìn)行文件合并過程中埠巨，MapTask 以分區(qū)為單位進(jìn)行合并，對于某個分區(qū)现拒，它將采用多輪遞歸合并的方式辣垒，每輪合并io.sort.factor（默認(rèn)10）個文件，并將產(chǎn)生的文件重新加入待合并列表中印蔬，對文件排序后勋桶，重復(fù)以上過程，直到最終得到一個大文件侥猬。

8）讓每個 MapTask 最終只生成一個數(shù)據(jù)文件例驹，可避免同時打開大量文件和同時讀取大量小文件產(chǎn)生的隨機(jī)讀取帶來的開銷。

3.5 ReduceTask 工作機(jī)制

image

1）Copy 階段：ReduceTask 從各個 MapTask 上遠(yuǎn)程拷貝一片數(shù)據(jù)退唠，并針對某一片數(shù)據(jù)鹃锈，如果其大小超過一定閾值，則寫到磁盤上瞧预，否則直接放到內(nèi)存中屎债。

2）Merge 階段：在遠(yuǎn)程拷貝數(shù)據(jù)的同時，ReduceTask 啟動了兩個后臺線程對內(nèi)存和磁盤上的文件進(jìn)行合并垢油，以防止內(nèi)存使用過多或磁盤上文件過多盆驹。

3）Sort 階段：按照 MapReduce 語義，用戶編寫 reduce() 函數(shù)輸入數(shù)據(jù)是按 key 進(jìn)行聚集的一組數(shù)據(jù)滩愁，為了將 key 相同的數(shù)據(jù)聚在一起躯喇，Hadoop 采用了基于排序的策略，由于各個 MapTask 已經(jīng)實現(xiàn)對自己的處理結(jié)果進(jìn)行了局部排序惊楼，因此玖瘸，ReduceTask 只需對所有數(shù)據(jù)進(jìn)行一次歸并排序即可。

4）Reduce 階段：reduce() 函數(shù)將計算結(jié)果寫到 HDFS 上檀咙。

ReduceTask 的并行度同樣影響整個 Job 的執(zhí)行并發(fā)度和執(zhí)行效率雅倒，但與 MapTask 的并發(fā)數(shù)由切片數(shù)決定不同，ReduceTask 數(shù)量的決定是可以直接手動設(shè)置：

job.setNumReduceTasks(4);

注意事項：

ReduceTask=0弧可，表示沒有 Reduce 階段蔑匣，輸出文件個數(shù)和 Map 個數(shù)一致
ReduceTask 默認(rèn)值就是 1，所以輸出文件個數(shù)為一個
如果數(shù)據(jù)分布不均勻棕诵，就有可能在 Reduce 階段產(chǎn)生數(shù)據(jù)傾斜
ReduceTask 數(shù)量并不是任意設(shè)置裁良，還要考慮業(yè)務(wù)邏輯需求，有些情況下校套，需要計算全局匯總結(jié)果价脾，就只能有 1 個ReduceTask
具體多少個 ReduceTask，需要根據(jù)集群性能而定
如果分區(qū)數(shù)不是 1笛匙，但是 ReduceTask 為 1侨把，不會執(zhí)行分區(qū)過程

3.6 OutputFormat 數(shù)據(jù)輸出

3.6.1 OutputFormat 接口實現(xiàn)類

image

3.6.2 自定義 OutputFormat

image

需求分析：

image

代碼實現(xiàn)：

# FilterOutputFormat
package com.djm.mapreduce.outputformat;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class FilterOutputFormat extends FileOutputFormat<text, nullwritable> {
    @Override
    public RecordWriter<text, nullwritable> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
        return new FilterRecordWriter(job);
    }
}

# FilterRecordWriter
package com.djm.mapreduce.outputformat;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;


public class FilterRecordWriter extends RecordWriter<text, nullwritable> {

    private FSDataOutputStream atguiguOut = null;
    private FSDataOutputStream otherOut = null;

    public FilterRecordWriter() {
    }

    public FilterRecordWriter(TaskAttemptContext job) {
        FileSystem fs;

        try {
            fs = FileSystem.get(job.getConfiguration());
            Path atguigu = new Path("C:\\Application\\Apache\\hadoop-2.7.2\\djm.log");
            Path other = new Path("C:\\Application\\Apache\\hadoop-2.7.2\\other.log");
            atguiguOut = fs.create(atguigu);
            otherOut = fs.create(other);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void write(Text key, NullWritable value) throws IOException, InterruptedException {
        if (key.toString().contains("atguigu")) {
            atguiguOut.write(key.toString().getBytes());
        } else {
            otherOut.write(key.toString().getBytes());
        }
    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        IOUtils.closeStream(atguiguOut);
        IOUtils.closeStream(otherOut);
    }
}


# FilterDriver
package com.djm.mapreduce.outputformat;

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

public class FilterDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Job job = Job.getInstance(new Configuration());
        job.setJarByClass(FilterDriver.class);
        job.setMapperClass(FilterMapper.class);
        job.setReducerClass(FilterReduce.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
        job.setOutputFormatClass(FilterOutputFormat.class);
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

3.7 Join

3.7.1 Reduce Join

工作原理：

Map 端

為來自不同表或文件的 key/value 對犀变，打標(biāo)簽以區(qū)別不同來源的記錄，然后用連接字段作為 key秋柄，其余部分和新加的標(biāo)志作為 value获枝，最后進(jìn)行輸出。
Reduce端

在 Reduce 端以連接字段作為 key 的分組已經(jīng)完成骇笔，我們只需要在每一個分組當(dāng)中將那些來源于不同文件的記錄分開省店，最后進(jìn)行合并。

需求分析：

image

代碼實現(xiàn)：

# TableBean
package com.djm.mapreduce.table;

import org.apache.hadoop.io.Writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

@Data
public class TableBean implements Writable {

    private String orderId;

    private String productId;

    private int amount;

    private String pname;

    private String flag;

    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(orderId);
        out.writeUTF(productId);
        out.writeInt(amount);
        out.writeUTF(pname);
        out.writeUTF(flag);

    }

    @Override
    public void readFields(DataInput in) throws IOException {
        this.orderId = in.readUTF();
        this.productId = in.readUTF();
        this.amount = in.readInt();
        this.pname = in.readUTF();
        this.flag = in.readUTF();

    }
}

# TableMapper
package com.djm.mapreduce.table;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class TableMapper extends Mapper<longwritable, text, tablebean>{

    String name;
    TableBean bean = new TableBean();
    Text k = new Text();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        FileSplit split = (FileSplit) context.getInputSplit();
        name = split.getPath().getName();
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        if (name.startsWith("order")) {// 訂單表處理
            String[] fields = line.split("\t");
            bean.setOrder_id(fields[0]);
            bean.setP_id(fields[1]);
            bean.setAmount(Integer.parseInt(fields[2]));
            bean.setPname("");
            bean.setFlag("order");
            k.set(fields[1]);
        }else {
            String[] fields = line.split("\t");
            bean.setP_id(fields[0]);
            bean.setPname(fields[1]);
            bean.setFlag("pd");
            bean.setAmount(0);
            bean.setOrder_id("");
            k.set(fields[0]);
        }
        context.write(k, bean);
    }
}

# TableReducer
package com.djm.mapreduce.table;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;

public class TableReducer extends Reducer<text, tablebean, nullwritable> {
    @Override
    protected void reduce(Text key, Iterable<tablebean> values, Context context) throws IOException, InterruptedException {
        ArrayList<tablebean> orderBeans = new ArrayList&lt;&gt;();
        TableBean pdBean = new TableBean();
        for (TableBean bean : values) {
            if ("order".equals(bean.getFlag())) {
                TableBean orderBean = new TableBean();
                try {
                    BeanUtils.copyProperties(orderBean, bean);
                } catch (IllegalAccessException | InvocationTargetException e) {
                    e.printStackTrace();
                }
                orderBeans.add(orderBean);
            }
            else {
                try {
                    BeanUtils.copyProperties(pdBean, bean);
                } catch (IllegalAccessException | InvocationTargetException e) {
                    e.printStackTrace();
                }
            }

        }
        for (TableBean bean :orderBeans) {
            bean.setPname (pdBean.getPname());
            context.write(bean, NullWritable.get());
        }
    }
}

package com.djm.mapreduce.table;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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;
import java.net.URI;
import java.net.URISyntaxException;

public class TableDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException {
        Job job = Job.getInstance(new Configuration());
        job.setJarByClass(TableDriver.class);
        job.setMapperClass(TableMapper.class);
        job.setReducerClass(TableReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(TableBean.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.addCacheFile(new URI("file:///C:/Application/Apache/hadoop-2.7.2/input/pd.txt"));
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

3.7.2 Map Join

Map Join 適用于一張表十分小笨触、一張表很大的場景懦傍。

優(yōu)點：

在 Map 端緩存多張表，提前處理業(yè)務(wù)邏輯旭旭，這樣增加 Map 端業(yè)務(wù)谎脯，減少 Reduce 端數(shù)據(jù)的壓力，就可以盡可能的減少數(shù)據(jù)傾斜持寄。

需求分析：

image

代碼實現(xiàn)：

# TableMapper
package com.djm.mapreduce.table;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

public class TableMapper extends Mapper<longwritable, text, nullwritable> {

    private Text k = new Text();

    private Map<string, string> pdMap = new HashMap&lt;&gt;();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        URI[] cacheFiles = context.getCacheFiles();
        String path = cacheFiles[0].getPath();

        BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(path), StandardCharsets.UTF_8));
        String line;
        while(StringUtils.isNotEmpty(line = reader.readLine())){
            String[] fields = line.split("\t");
            pdMap.put(fields[0], fields[1]);
        }
        reader.close();
    }

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

        String[] fields = value.toString().split("\t");
        String pId = fields[1];
        String pdName = pdMap.get(pId);
        k.set(fields[0] + "\t"+ pdName + "\t" + fields[2]);
        context.write(k, NullWritable.get());
    }
}

# TableDriver
package com.djm.mapreduce.table;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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;
import java.net.URI;
import java.net.URISyntaxException;

public class TableDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException, URISyntaxException {
        Job job = Job.getInstance(new Configuration());
        job.setJarByClass(TableDriver.class);
        job.setMapperClass(TableMapper.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.addCacheFile(new URI("file:///C:/Application/Apache/hadoop-2.7.2/input/pd.txt"));
        job.setNumReduceTasks(0);
        boolean result = job.waitForCompletion(true);
        System.exit(result ? 0 : 1);
    }
}

3.8 ETL

在運行核心業(yè)務(wù) MapReduce 程序之前源梭，往往要先對數(shù)據(jù)進(jìn)行清洗，清理掉不符合用戶要求的數(shù)據(jù)稍味。清理的過程往往只需要運行 Mapper 程序废麻，不需要運行 Reduce 程序。

需求分析：

需要在 Map 階段對輸入的數(shù)據(jù)根據(jù)規(guī)則進(jìn)行過濾清洗模庐。

代碼實現(xiàn)：

# LogBean
package com.djm.mapreduce.etl;

@Data
public class LogBean {

    private String remoteAddr;

    private String remoteUser;

    private String timeLocal;

    private String request;

    private String status;

    private String bodyBytesSent;

    private String httpReferer;

    private String httpUserAgent;

    private boolean valid = true;
}
# LogMapper
package com.djm.mapreduce.etl;

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

import java.io.IOException;

public class LogMapper extends Mapper<longwritable, text, nullwritable> {

    private Text k = new Text();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        LogBean bean = parseLog(line);
        if (!bean.isValid()) {
            return;
        }
        k.set(bean.toString());
        context.write(k, NullWritable.get());
    }

    private LogBean parseLog(String line) {
        LogBean logBean = new LogBean();
        String[] fields = line.split(" ");
        if (fields.length &gt; 11) {
            logBean.setRemoteAddr(fields[0]);
            logBean.setRemoteUser(fields[1]);
            logBean.setTimeLocal(fields[3].substring(1));
            logBean.setRequest(fields[6]);
            logBean.setStatus(fields[8]);
            logBean.setBodyBytesSent(fields[9]);
            logBean.setHttpReferer(fields[10]);
            if (fields.length &gt; 12) {
                logBean.setHttpUserAgent(fields[11] + " " + fields[12]);
            } else {
                logBean.setHttpUserAgent(fields[11]);
            }
            if (Integer.parseInt(logBean.getStatus()) &gt;= 400) {
                logBean.setValid(false);
            }
        } else {
            logBean.setValid(false);
        }
        return logBean;
    }
}

# LogDriver
package com.djm.mapreduce.etl;

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

public class LogDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Job job = Job.getInstance(new Configuration());
        job.setJarByClass(LogDriver.class);
        job.setMapperClass(LogMapper.class);
        job.setNumReduceTasks(0);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        job.waitForCompletion(true);
    }
}

3.9 MapReduce 開發(fā)總結(jié)

在編寫 MapReduce 程序時烛愧，需要考慮如下幾個方面：

Mapper

用戶根據(jù)業(yè)務(wù)需求實現(xiàn)其中三個方法：map() setup() cleanup ()

Partitioner分區(qū)

有默認(rèn)實現(xiàn) HashPartitioner，邏輯是根據(jù) key 的哈希值和 numReduces 來返回一個分區(qū)號
```
key.hashCode()&amp;Integer.MAXVALUE % numReduces
```
如果業(yè)務(wù)上有特別的需求掂碱，可以自定義分區(qū)

Comparable

當(dāng)我們用自定義的對象作為 key 來輸出時怜姿，就必須要實現(xiàn) WritableComparable 接口，重寫其中的 compareTo() 方法
部分排序：對最終輸出的每一個文件進(jìn)行內(nèi)部排序
全排序：對所有數(shù)據(jù)進(jìn)行排序疼燥，通常只有一個 Reduce
二次排序：排序的條件有兩個

Combiner

Combiner 合并可以提高程序執(zhí)行效率沧卢，減少 IO 傳輸，但是使用時必須不能影響原有的業(yè)務(wù)處理結(jié)果

GroupingComparator

在 Reduce 端對 key 進(jìn)行分組

Reducer

用戶根據(jù)業(yè)務(wù)需求實現(xiàn)其中三個方法：reduce() setup() cleanup ()

OutputFormat

默認(rèn)實現(xiàn)類是 TextOutputFormat醉者，功能邏輯是：將每一個 KV 對但狭，向目標(biāo)文本文件輸出一行
將 SequenceFileOutputFormat 輸出作為后續(xù) MapReduce任務(wù)的輸入，這便是一種好的輸出格式撬即，因為它的格式緊湊立磁，很容易被壓縮
用戶還可以自定義 OutputFormat

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者

人面猴
序言：七十年代末，一起剝皮案震驚了整個濱河市剥槐，隨后出現(xiàn)的幾起案子唱歧，更是在濱河造成了極大的恐慌，老刑警劉巖粒竖，帶你破解...
沈念sama閱讀 212,816評論 6贊 492
死咒
序言：濱河連續(xù)發(fā)生了三起死亡事件颅崩，死亡現(xiàn)場離奇詭異绍刮，居然都是意外死亡，警方通過查閱死者的電腦和手機(jī)挨摸，發(fā)現(xiàn)死者居然都...
沈念sama閱讀 90,729評論 3贊 385
救了他兩次的神仙讓他今天三更去死
文/潘曉璐我一進(jìn)店門，熙熙樓的掌柜王于貴愁眉苦臉地迎上來岁歉，“玉大人得运，你說我怎么就攤上這事」疲” “怎么了熔掺？”我有些...
開封第一講書人閱讀 158,300評論 0贊 348
道士緝兇錄：失蹤的賣姜人
文/不壞的土叔我叫張陵，是天一觀的道長非剃。經(jīng)常有香客問我置逻，道長，這世上最難降的妖魔是什么备绽？我笑而不...
開封第一講書人閱讀 56,780評論 1贊 285
?港島之戀（遺憾婚禮）
正文為了忘掉前任券坞，我火速辦了婚禮，結(jié)果婚禮上肺素，老公的妹妹穿的比我還像新娘恨锚。我一直安慰自己，他們只是感情好倍靡，可當(dāng)我...
茶點故事閱讀 65,890評論 6贊 385
惡毒庶女頂嫁案：這布局不是一般人想出來的
文/花漫我一把揭開白布猴伶。她就那樣靜靜地躺著，像睡著了一般塌西。火紅的嫁衣襯著肌膚如雪他挎。梳的紋絲不亂的頭發(fā)上，一...
開封第一講書人閱讀 50,084評論 1贊 291
城市分裂傳說
那天捡需，我揣著相機(jī)與錄音办桨，去河邊找鬼。笑死栖忠，一個胖子當(dāng)著我的面吹牛崔挖，可吹牛的內(nèi)容都是我干的。我是一名探鬼主播庵寞，決...
沈念sama閱讀 39,151評論 3贊 410
雙鴛鴦連環(huán)套：你想象不到人心有多黑
文/蒼蘭香墨我猛地睜開眼狸相，長吁一口氣：“原來是場噩夢啊……” “哼！你這毒婦竟也來了捐川？” 一聲冷哼從身側(cè)響起脓鹃，我...
開封第一講書人閱讀 37,912評論 0贊 268
萬榮殺人案實錄
序言：老撾萬榮一對情侶失蹤，失蹤者是張志新（化名）和其女友劉穎古沥，沒想到半個月后瘸右，有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體娇跟，經(jīng)...
沈念sama閱讀 44,355評論 1贊 303
?護(hù)林員之死
正文獨居荒郊野嶺守林人離奇死亡，尸身上長有42處帶血的膿包…… 初始之章·張勛以下內(nèi)容為張勛視角年9月15日...
茶點故事閱讀 36,666評論 2贊 327
?白月光啟示錄
正文我和宋清朗相戀三年太颤，在試婚紗的時候發(fā)現(xiàn)自己被綠了苞俘。大學(xué)時的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片。...
茶點故事閱讀 38,809評論 1贊 341
活死人
序言：一個原本活蹦亂跳的男人離奇死亡龄章，死狀恐怖吃谣，靈堂內(nèi)的尸體忽然破棺而出，到底是詐尸還是另有隱情做裙，我是刑警寧澤岗憋，帶...
沈念sama閱讀 34,504評論 4贊 334
?日本核電站爆炸內(nèi)幕
正文年R本政府宣布，位于F島的核電站，受9級特大地震影響，放射性物質(zhì)發(fā)生泄漏推姻。R本人自食惡果不足惜，卻給世界環(huán)境...
茶點故事閱讀 40,150評論 3贊 317
男人毒藥：我在死后第九天來索命
文/蒙蒙一监徘、第九天我趴在偏房一處隱蔽的房頂上張望。院中可真熱鬧吧碾，春花似錦耐量、人聲如沸。這莊子的主人今日做“春日...
開封第一講書人閱讀 30,882評論 0贊 21
一樁弒父案廊蜒，背后竟有這般陰謀
文/蒼蘭香墨我抬頭看了看天上的太陽。三九已至溅漾，卻和暖如春山叮，著一層夾襖步出監(jiān)牢的瞬間，已是汗流浹背添履。一陣腳步聲響...
開封第一講書人閱讀 32,121評論 1贊 267
情欲美人皮
我被黑心中介騙來泰國打工屁倔，沒想到剛下飛機(jī)就差點兒被人妖公主榨干…… 1. 我叫王不留，地道東北人暮胧。一個月前我還...
沈念sama閱讀 46,628評論 2贊 362
代替公主和親
正文我出身青樓锐借，卻偏偏與公主長得像，于是被迫代替她去往敵國和親往衷。傳聞我的和親對象是個殘疾皇子钞翔，可洞房花燭夜當(dāng)晚...
茶點故事閱讀 43,724評論 2贊 351

Hadoop 之 MapReduce

1 MapReduce 概述

1.1 MapReduce 優(yōu)缺點

1.2 MapReduce 核心思想

1.3 MapReduce 進(jìn)程

1.4 常用數(shù)據(jù)序列化類型

1.5 MapReduce 編程規(guī)范

1.6 WordCount 案例實操

2 Hadoop 序列化

2.1 為什么不使用 Java 序列化框架進(jìn)行序列化

2.2 自定義 bean 對象實現(xiàn)序列化接口

2.3 序列化案例實操

3 MapReduce 框架原理

3.1 InputFormat 數(shù)據(jù)輸入

3.1.1 切片與MapTask并行度決定機(jī)制

3.1.2 FileInputFormat 切片機(jī)制

3.1.3 CombineTextInputFormat 切片機(jī)制

3.1.4 FileInputFormat 的其他實現(xiàn)類

3.1.5 自定義 InputFormat

3.2 MapReduce工作流程

3.3 Shuffle 機(jī)制

3.3.1 Partition 分區(qū)

3.3.2 WritableComparable 排序

3.3.3 GroupingComparator 分組

3.4 MapTask 工作機(jī)制

3.5 ReduceTask 工作機(jī)制

3.6 OutputFormat 數(shù)據(jù)輸出

3.6.1 OutputFormat 接口實現(xiàn)類

3.6.2 自定義 OutputFormat

3.7 Join

3.7.1 Reduce Join

3.7.2 Map Join

3.8 ETL

3.9 MapReduce 開發(fā)總結(jié)

推薦閱讀更多精彩內(nèi)容