Hadoop之——MapReduce實戰（一）

阿新 • • 發佈：2019-01-15

MapReduce概述

MapReduce是一種分散式計算模型，由Google提出，主要用於搜尋領域，解決海量資料的計算問題.

MR由兩個階段組成：Map和Reduce，使用者只需要實現map()和reduce()兩個函式，即可實現分散式計算，非常簡單。

這兩個函式的形參是key、value對，表示函式的輸入資訊。

MR執行流程

MapReduce原理

執行步驟

1. map任務處理

1.1 讀取輸入檔案內容，解析成key、value對。對輸入檔案的每一行，解析成key、value對。每一個鍵值對呼叫一次map函式。

1.2 寫自己的邏輯，對輸入的key、value處理，轉換成新的key、value輸出。

1.3 對輸出的key、value進行分割槽。

1.4 對不同分割槽的資料，按照key進行排序、分組。相同key的value放到一個集合中。

1.5 (可選)分組後的資料進行歸約。

2.reduce任務處理

2.1 對多個map任務的輸出，按照不同的分割槽，通過網路copy到不同的reduce節點。

2.2 對多個map任務的輸出進行合併、排序。寫reduce函式自己的邏輯，對輸入的key、value處理，轉換成新的key、value輸出。

2.3 把reduce的輸出儲存到檔案中。

例子：實現WordCountApp

map、reduce鍵值對格式

WordCountApp的驅動程式碼

public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();    //載入配置檔案
    Job job = new Job(conf);    //建立一個job，供JobTracker使用
    job.setJarByClass(WordCountApp.class);
		
    job.setMapperClass(WordCountMapper.class);
    job.setReducerClass(WordCountReducer.class);
    FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.1.10:9000/input"));
    FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.1.10:9000/output"));
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
		
    job.waitForCompletion(true);
}

JobTracker

負責接收使用者提交的作業，負責啟動、跟蹤任務執行。

JobSubmissionProtocol是JobClient與JobTracker通訊的介面。

InterTrackerProtocol是TaskTracker與JobTracker通訊的介面。

TaskTracker

負責執行任務

JobClient

是使用者作業與JobTracker互動的主要介面。

負責提交作業的，負責啟動、跟蹤任務執行、訪問任務狀態和日誌等

最小的MapReduce驅動

Configuration configuration = new Configuration();
Job job = new Job(configuration, "HelloWorld");
job.setInputFormat(TextInputFormat.class);
job.setMapperClass(IdentityMapper.class);
job.setMapOutputKeyClass(LongWritable.class);
job.setMapOutputValueClass(Text.class);
job.setPartitionerClass(HashPartitioner.class);
job.setNumReduceTasks(1);
job.setReducerClass(IdentityReducer.class);
job.setOutputKeyClass(LongWritable.class);
job.setOutputValueClass(Text.class);
job.setOutputFormat(TextOutputFormat.class);
job.waitForCompletion(true);

MapReduce驅動預設的設定

序列化概念

序列化（Serialization）是指把結構化物件轉化為位元組流。

反序列化（Deserialization）是序列化的逆過程。即把位元組流轉回結構化物件。

Java序列化（java.io.Serializable

Hadoop序列化的特點

序列化格式特點：

緊湊：高效使用儲存空間。
快速：讀寫資料的額外開銷小
可擴充套件：可透明地讀取老格式的資料

互操作：支援多語言的互動

Hadoop的序列化格式：Writable

注意：Java序列化的不足：

1.不精簡。附加資訊多。不大適合隨機訪問。

2.儲存空間大。遞迴地輸出類的超類描述直到不再有超類。序列化圖物件，反序列化時為每個物件新建一個例項。相反。Writable物件可以重用。

3.擴充套件性差。而Writable方便使用者自定義

Hadoop序列化的作用

序列化在分散式環境的兩大作用：程序間通訊，永久儲存。

Hadoop節點間通訊。

Writable介面

Writable介面, 是根據 DataInput和 DataOutput 實現的簡單、有效的序列化物件.

MR的任意Key和Value必須實現Writable介面.

MR的任意key必須實現WritableComparable介面

常用的Writable實現類

Text一般認為它等價於java.lang.String的Writable。針對UTF-8序列。

例:

Text test = new Text("test");

IntWritable one = new IntWritable(1);

Writable

① write 是把每個物件序列化到輸出流

② readFields是把輸入流位元組反序列化

③ 實現WritableComparable.

④ Java值物件的比較：一般需要重寫toString(),hashCode(),equals()方法

基於檔案的儲存結構

SequenceFile 無序儲存

MapFile 會對key建立索引檔案，value按key順序儲存

基於MapFile的結構有:

ArrayFile 像我們使用的陣列一樣，key值為序列化的數字

SetFile 他只有key，value為不可變的資料

BloomMapFile 在 MapFile 的基礎上增加了一個 /bloom檔案，包含的是二進位制的過濾表，在每一次寫操作完成時，會更新這個過濾表。

MapReduce的輸入處理類

FileInputFormat:
FileInputFormat是所有以檔案作為資料來源的InputFormat實現的基類，FileInputFormat儲存作為job輸入的所有檔案，並實現了對輸入檔案計算splits的方法。至於獲得記錄的方法是有不同的子類——TextInputFormat進行實現的。

InputFormat

InputFormat 負責處理MR的輸入部分.

有三個作用:

驗證作業的輸入是否規範.

把輸入檔案切分成InputSplit.

提供RecordReader 的實現類，把InputSplit讀到Mapper中進行處理

InputSplit

在執行mapreduce之前，原始資料被分割成若干split，每個split作為一個map任務的輸入，在map執行過程中split會被分解成一個個記錄（key-value對），map會依次處理每一個記錄。

FileInputFormat只劃分比HDFS block大的檔案，所以FileInputFormat劃分的結果是這個檔案或者是這個檔案中的一部分.

如果一個檔案的大小比block小，將不會被劃分，這也是Hadoop處理大檔案的效率要比處理很多小檔案的效率高的原因。

當Hadoop處理很多小檔案（檔案大小小於hdfs block大小）的時候，由於FileInputFormat不會對小檔案進行劃分，所以每一個小檔案都會被當做一個split並分配一個map任務，導致效率底下。

例如：一個1G的檔案，會被劃分成16個64MB的split，並分配16個map任務處理，而10000個100kb的檔案會被10000個map任務處理。

TextInputFormat

TextInputformat是預設的處理類，處理普通文字檔案。

檔案中每一行作為一個記錄，他將每一行在檔案中的起始偏移量作為key，每一行的內容作為value。

預設以\n或回車鍵作為一行記錄。

TextInputFormat繼承了FileInputFormat。

InputFormat類的層次結構

其他輸入類

CombineFileInputFormat

相對於大量的小檔案來說，hadoop更合適處理少量的大檔案。

CombineFileInputFormat可以緩解這個問題，它是針對小檔案而設計的。

KeyValueTextInputFormat

當輸入資料的每一行是兩列，並用tab分離的形式的時候，KeyValueTextInputformat處理這種格式的檔案非常適合。

NLineInputformat
NLineInputformat可以控制在每個split中資料的行數。

SequenceFileInputformat

當輸入檔案格式是sequencefile的時候，要使用SequenceFileInputformat作為輸入

自定義輸入格式

1）繼承FileInputFormat基類。

2）重寫裡面的getSplits(JobContextcontext)方法。

3）重寫createRecordReader(InputSplitsplit, TaskAttemptContext context)方法。

Hadoop的輸出

TextOutputformat

預設的輸出格式，key和value中間值用tab隔開的。

SequenceFileOutputformat

將key和value以sequencefile格式輸出。

SequenceFileAsOutputFormat

將key和value以原始二進位制的格式輸出。

MapFileOutputFormat

將key和value寫入MapFile中。由於MapFile中的key是有序的，所以寫入的時候必須保證記錄是按key值順序寫入的。

MultipleOutputFormat

預設情況下一個reducer會產生一個輸出，但是有些時候我們想一個reducer產生多個輸出，MultipleOutputFormat和MultipleOutputs可以實現這個功能。