1 MapRedcue的介紹

1.1 MapReduce定義

MapReduce是一個進行分布式運算的編程框架，使用戶開發(fā)基于hadoop進行數(shù)據(jù)分析的核心框架。 MapReduce 核心功能就是將用戶編寫的業(yè)務(wù)邏輯代碼和自帶的默認組件整合成一個完整的 分布式運算程序，并發(fā)運行在一個 Hadoop 集群上。

1.2 MapReduce的思想

MapReduce的思想核心是分而治之，適用于大規(guī)模數(shù)據(jù)處理場景。

map負責分，將復雜的任務(wù)拆解成可以并行計算的若干個任務(wù)來處理

reduce負責合，對map階段的結(jié)果進行全局匯總

比如說：老師作業(yè)留的有點多，一個人寫太費勁了，就可以用MapReduce這種分而治之的思想，將作業(yè)進行map處理，分給不同的人，最后所有寫完的部分發(fā)到群里進行reduce匯總，復雜的作業(yè)簡簡單單。

1.3 MapReduce優(yōu)點

易于編程

MapReduce將做什么和怎么做分開了，提供了一些接口，程序員只需關(guān)注應(yīng)用層上的問題。具體如何實現(xiàn)并行計算任務(wù)則被隱藏了起來。

擴展性

當計算資源不足時，可以增加機器來提高擴展能力

高容錯

一臺機器掛了，可以將計算任務(wù)轉(zhuǎn)移到另一臺節(jié)點上進行

適合PB級海量數(shù)據(jù)的離線處理

1.4 MapReduce的缺點

不擅長實時計算

無法做到在毫秒級別返回結(jié)果

不擅長流式計算

MapReduce處理的數(shù)據(jù)源只能是靜態(tài)的，不能動態(tài)變化

不擅長DAG(有向無環(huán)圖)計算

每個MR作業(yè)處理結(jié)束，結(jié)果都會寫入到磁盤，造成大量的磁盤IO，導致性能低下

1.5 MapReduce進程

一個MapReduce程序在分布式運行時有三類的實例進程

• MrAppMaster : 負責整個程序的過程調(diào)度及狀態(tài)協(xié)調(diào)
• MapTask : 負責Map階段的數(shù)據(jù)處理流程
• ReduceTask ： 負責Reduce階段的數(shù)據(jù)處理流程

1.6 MapReduce-WordCount

import java.io.IOException;
import java.util.StringTokenizer;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class WordCount {
  public static class TokenizerMapper
       extends Mapper<Object, Text, Text, IntWritable>{
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();
    public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }
  }
  public static class IntSumReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    private IntWritable result = new IntWritable();
    public void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }
  }
  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    Job job = Job.getInstance(conf, "word count");
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

以上代碼實現(xiàn)了兩個類：TokenizerMapper和IntSumReducer，它們分別實現(xiàn)了Map和Reduce功能。

Map函數(shù)將輸入的每一行文本進行分詞，并將每個單詞映射為一個鍵值對，其中鍵為單詞，值為1，然后將這些鍵值對輸出給Reduce函數(shù)。

Reduce函數(shù)將相同鍵的值相加，并將最終結(jié)果輸出。

在這個例子中，CombinerClass被設(shè)置為相同的Reducer類，用于在Map任務(wù)結(jié)束后本地合并中間結(jié)果，以減少網(wǎng)絡(luò)傳輸。

最后，將輸入文件和輸出文件的路徑作為命令行參數(shù)傳遞給main函數(shù)，并啟動MapReduce作業(yè)。

• job的講解

在Hadoop MapReduce程序中，Job對象是用來定義和運行一個MapReduce作業(yè)的。

Job對象的主要功能是封裝了整個MapReduce作業(yè)的配置和運行信息，包括輸入數(shù)據(jù)和輸出數(shù)據(jù)的路徑、Mapper類和Reducer類的設(shè)置、中間結(jié)果的輸出類型和格式、作業(yè)的提交方式等。

在main函數(shù)中，我們創(chuàng)建一個Job對象并設(shè)置它的相關(guān)屬性。

Job.getInstance()方法返回一個新的Job實例，其中的Configuration對象用來指定作業(yè)的一些配置信息。

setJarByClass()方法用來設(shè)置作業(yè)的jar包，它的參數(shù)是定義MapReduce作業(yè)的主類。

setMapperClass()、setCombinerClass()和setReducerClass()方法用來指定Mapper、Combiner和Reducer的實現(xiàn)類。

setOutputKeyClass()和setOutputValueClass()方法分別用來設(shè)置MapReduce作業(yè)的輸出鍵和輸出值的類型。

FileInputFormat.addInputPath()和FileOutputFormat.setOutputPath()方法用來指定輸入文件和輸出文件的路徑。

最后，我們調(diào)用job.waitForCompletion()方法來提交并運行作業(yè)，并等待作業(yè)完成。

如果作業(yè)成功完成，waitForCompletion()方法將返回true，否則返回false。

如果作業(yè)失敗，我們可以通過job.getJobState()方法來獲取作業(yè)的狀態(tài)信息，或者查看作業(yè)的日志信息來進行排錯和調(diào)試。

2 Hadoop序列化

2.1 序列化的定義

序列化就是將內(nèi)存中對象轉(zhuǎn)換成字節(jié)序列，便于存儲到磁盤和網(wǎng)絡(luò)傳輸

反序列化時將字節(jié)序列或磁盤中的持久化數(shù)據(jù)轉(zhuǎn)換成內(nèi)存中的對象

一般來說，對象只能在本地進程中使用，不能通過網(wǎng)絡(luò)發(fā)送到另一臺計算機

序列化可以存儲對象，可以將對象發(fā)送到遠程計算機

2.2 hadoop序列化和java序列化的區(qū)別

Hadoop序列化和Java序列化都是將對象轉(zhuǎn)換為字節(jié)序列以便于在網(wǎng)絡(luò)上傳輸或者存儲到磁盤等持久化存儲介質(zhì)中。它們的主要區(qū)別在于以下幾點：

• 序列化速度和效率不同：Hadoop序列化比Java序列化更快，因為它采用的是二進制格式，而Java序列化采用的是基于文本的XML或JSON格式。
• 支持的數(shù)據(jù)類型不同：Hadoop序列化支持的數(shù)據(jù)類型比Java序列化更多，包括基本類型、數(shù)組、集合、映射、枚舉、自定義類等。
• 序列化后的數(shù)據(jù)大小不同：Hadoop序列化生成的字節(jié)流比Java序列化生成的字節(jié)流更小，因為它使用更緊湊的二進制格式，這對于在網(wǎng)絡(luò)上傳輸和存儲到磁盤等介質(zhì)中非常重要。
• 可移植性不同：Java序列化生成的字節(jié)流只能被Java程序讀取，而Hadoop序列化生成的字節(jié)流可以被任何語言的程序讀取，因為它使用了通用的二進制格式。

總的來說，Hadoop序列化更適合用于大規(guī)模數(shù)據(jù)的處理和分布式計算，而Java序列化更適合用于小規(guī)模數(shù)據(jù)的傳輸和存儲。

3 MapReduce 的原理

3.1 MapReduce 工作的過程

數(shù)據(jù)輸入：MapReduce從Hadoop分布式文件系統(tǒng)（HDFS）中讀取輸入數(shù)據(jù)，并將其分成固定大小的數(shù)據(jù)塊，每個數(shù)據(jù)塊大小通常為64MB。

Map階段：在Map階段，MapReduce將每個數(shù)據(jù)塊分發(fā)給一組可擴展的計算節(jié)點，每個計算節(jié)點運行Map函數(shù)來處理它們分配的數(shù)據(jù)塊。Map函數(shù)將輸入數(shù)據(jù)轉(zhuǎn)換為一組鍵值對（Key-Value Pairs）的形式，這些鍵值對可以被后續(xù)的Reduce函數(shù)處理。

Shuffle階段：在Map函數(shù)處理完數(shù)據(jù)之后，MapReduce框架將所有的鍵值對按照它們的鍵進行排序，并將相同鍵的值合并在一起。這個過程通常被稱為“Shuffle”。

Reduce階段：在Reduce階段，MapReduce框架將合并后的鍵值對發(fā)送到一組可擴展的計算節(jié)點。每個節(jié)點運行Reduce函數(shù)來處理它們收到的所有鍵值對，并生成最終的輸出結(jié)果。

數(shù)據(jù)輸出：在Reduce函數(shù)處理完數(shù)據(jù)之后，MapReduce將輸出結(jié)果寫入HDFS中。

這些步驟中，Map和Reduce函數(shù)是由開發(fā)者自行編寫的，它們實現(xiàn)了具體的業(yè)務(wù)邏輯。MapReduce框架提供了分布式計算的基礎(chǔ)設(shè)施，負責管理計算節(jié)點、任務(wù)分配、故障處理等任務(wù)，以保證整個計算過程的可靠性和高效性。

總的來說，MapReduce框架的原理是將大數(shù)據(jù)集劃分成多個小數(shù)據(jù)塊，然后將這些數(shù)據(jù)塊分發(fā)給多個計算節(jié)點并行處理，最后將處理結(jié)果合并為一個最終結(jié)果。它通過這種方式來充分利用集群中的計算資源，提高計算效率和數(shù)據(jù)處理能力。

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

3.2.1 切片

數(shù)據(jù)塊：Block時HDFS在物理上對數(shù)據(jù)進行切塊，是HDFS存儲數(shù)據(jù)的單位

數(shù)據(jù)切片：數(shù)據(jù)切片是在邏輯上對輸入進行切片。切片是MR程序計算輸入數(shù)據(jù)的單位，一個切片會啟動一個MapTask

客戶端提交job時的切片數(shù)決定了map階段的并行度

默認情況下，切片大小為BlockSize

切片不會考慮數(shù)據(jù)整體，是對每個文件進行單獨切片

3.2.2 FileInputFormat

在MapReduce中，F(xiàn)ileInputFormat是一個抽象類，用于定義如何將文件分割成輸入數(shù)據(jù)塊并生成適合Mapper處理的RecordReader。它是MapReduce中的輸入格式類之一，用于讀取Hadoop分布式文件系統(tǒng)（HDFS）或本地文件系統(tǒng)中的數(shù)據(jù)。

FileInputFormat包括兩個關(guān)鍵方法：getSplits()和createRecordReader()。

getSplits()方法將輸入文件劃分成適合Map任務(wù)的數(shù)據(jù)塊，每個數(shù)據(jù)塊對應(yīng)一個Map任務(wù)。該方法返回一個InputSplit對象的數(shù)組，其中每個InputSplit表示一個文件數(shù)據(jù)塊。

createRecordReader()方法創(chuàng)建一個RecordReader對象，用于讀取InputSplit中的數(shù)據(jù)塊。RecordReader負責讀取一個數(shù)據(jù)塊中的所有記錄，并將它們轉(zhuǎn)換成key-value對。

FileInputFormat還提供了一些其他的方法，如isSplitable()用于判斷一個文件是否可以被劃分成多個數(shù)據(jù)塊。

Hadoop提供了一些預定義的FileInputFormat類，如TextInputFormat用于讀取文本文件，SequenceFileInputFormat用于讀取SequenceFile格式的文件等，用戶也可以通過繼承FileInputFormat自定義輸入格式類。

3.2.3 TextInputFormat

TextInputFormat是FileInputFormat默認的實現(xiàn)類，按行讀取每條記錄

key為該行的起始字節(jié)偏移量，為LongWritable類型

value 為這一行的內(nèi)容，不包括終止符，為Text類型

3.2.4 CombineTextInputFormat

TextInputFormat是按文件進行規(guī)劃分片，不管文件有多小，都是是一個單獨的切片，這樣會產(chǎn)生大量的MapTask，效率低下

CombineTextInputFormat用于小文件過多的場景，可以將多個小文件在邏輯上劃分到一個切片

決定哪些塊放入同一個分片時，CombineTextInputFormat會考慮到節(jié)點和機架的因素，所以在MR作業(yè)處理輸入的速度不會下降

CombineTextInputFormat不僅可以很好的處理小文件，在處理大文件時也有好處，因為它在每個節(jié)點生成了一個分片，分片可能又多個塊組成，CombineTextInputFormat使map操作中處理的數(shù)據(jù)量和HDFS中文件塊的大小的耦合度降低了

3.3 MapReduce工作機制

3.3.1 MapTask工作機制

• 讀取輸入數(shù)據(jù)：MapTask通過InputFormat獲得RecordReader，從輸入InputSplit中解析出KV
• Map階段：將解析出的KV交給map()函數(shù)處理，產(chǎn)生一系列新的KV
• Collect收集：數(shù)據(jù)處理完成之后，會調(diào)用OutputCollector.collect()輸出結(jié)果。在該函數(shù)的內(nèi)部，會生成KV分區(qū)，寫入環(huán)形緩沖區(qū)中
• Spill階段：環(huán)形緩沖區(qū)滿了之后，MR會將數(shù)據(jù)寫到本地磁盤，形成一個臨時文件，在寫入之前，會對數(shù)據(jù)進行一次排序
• merge階段：所有數(shù)據(jù)處理完畢之后，MapTask會對所有臨時文件進行一次合并，確保只生成一個數(shù)據(jù)文件

Spill階段詳情：

• 通過快速排序?qū)Νh(huán)形緩沖區(qū)內(nèi)的數(shù)據(jù)進行排序，先按照partition(后面會介紹)編號進行排序，然后再按照K進行排序。排序過后，數(shù)據(jù)以分區(qū)為單位聚集，分區(qū)內(nèi)的所有數(shù)按照K有序
• 按照分區(qū)編號由小到大將分區(qū)數(shù)據(jù)寫入工作目錄下的臨時文件 output/spillN.out(N表示當前溢寫的次數(shù))，如果設(shè)置了combiner(后面會介紹)，則寫入文件之前，還會將分區(qū)中的數(shù)據(jù)進行一次聚集操作
• 將分區(qū)數(shù)據(jù)的元數(shù)據(jù)寫入到內(nèi)存索引數(shù)據(jù)結(jié)構(gòu)SpillRecord中，每個分區(qū)的元數(shù)據(jù)包括臨時文件的偏移量、壓縮前后的數(shù)據(jù)大小，如果內(nèi)存索引大于1MB，會將內(nèi)存索引寫到文件 output.spillN.out.index中

3.3.2 Partition分區(qū)

在MapReduce計算模型中，Map任務(wù)會將生成的鍵值對按照鍵進行排序，并將其劃分到不同的分區(qū)中。分區(qū)的數(shù)量通常等于Reduce任務(wù)的數(shù)量。具體來說，Map任務(wù)會按照Partitioner函數(shù)定義的分區(qū)規(guī)則對鍵值對進行劃分。Partitioner函數(shù)將每個鍵值對映射到一個分區(qū)編號，然后Map任務(wù)將其輸出到對應(yīng)的分區(qū)中。

Partitioner函數(shù)通常是由用戶自定義實現(xiàn)的，其作用是將鍵值對映射到一個特定的分區(qū)。Hadoop提供了默認的Partitioner實現(xiàn)，即HashPartitioner，它將鍵哈希后取模得到分區(qū)編號，從而實現(xiàn)對鍵值對的劃分。在實際應(yīng)用中，用戶可以根據(jù)自己的需求自定義Partitioner函數(shù)，以便將鍵值對劃分到特定的分區(qū)中。

如果ReduceTask數(shù)量 > Partition數(shù)量，會產(chǎn)生多個空的輸出文件

如果ReduceTask數(shù)量 < Partition數(shù)量，會導致有分區(qū)的數(shù)據(jù)無處安放，會Exception

如果ReduceTask數(shù)量 = 1，則不管有多少個分區(qū)文件，最終都會只產(chǎn)生一個文件

3.3.3 Combiner合并

Combiner在每個mapTask所在的節(jié)點運行

Combiner對每個MapTask的輸出進行局部匯總，減少reduce階段的負擔

Combiner使用的前提是不能影響業(yè)務(wù)邏輯

3.3.4 ReduceTask工作機制

Copy：拉取數(shù)據(jù)，Reduce進程啟動copy進程(Fetcher)，通過HTTP的方式請求maptask獲取自己的文件，map task分區(qū)表示每個map task屬于哪個reduce task

Merge：ReduceTask啟動兩個線程對內(nèi)存和磁盤中的文件進行合并，防止文件過多。當內(nèi)存中的數(shù)據(jù)達到一定閾值時，就會啟動內(nèi)存到磁盤的merge，與map端的相似。直到?jīng)]有map端的數(shù)據(jù)才結(jié)束

合并排序：將數(shù)據(jù)合并成一個大數(shù)據(jù)，并進行排序

對排序后的數(shù)據(jù)調(diào)用reduce方法：對鍵相同的鍵值對調(diào)用reduce方法，每次調(diào)用會產(chǎn)生零個或多個鍵值對，最后將輸出的鍵值對存入HDFS。

3.3.5 shuffle機制

Shuffle階段的過程可以分為三個階段：

Map端的輸出：Map任務(wù)將生成的鍵值對按照鍵排序，并將其劃分到不同的分區(qū)中。如果Map任務(wù)的輸出緩存區(qū)已滿，則需要將其溢出到本地磁盤的臨時文件中。

數(shù)據(jù)傳輸：在Shuffle階段中，Map任務(wù)的輸出需要傳輸?shù)絉educe任務(wù)所在的節(jié)點，以便Reduce任務(wù)可以從中提取和合并數(shù)據(jù)。數(shù)據(jù)傳輸是Shuffle階段的關(guān)鍵步驟，其速度和效率直接影響整個MapReduce作業(yè)的性能。

Reduce端的輸入：Reduce任務(wù)需要從本地磁盤讀取屬于自己的分區(qū)的臨時文件，并對同一個分區(qū)中的鍵值對進行合并和排序。Reduce任務(wù)將合并后的結(jié)果輸出到最終的輸出文件中。

Shuffle階段是MapReduce計算模型中非常重要的一個階段，它的性能和效率對整個作業(yè)的執(zhí)行時間和性能影響非常大。因此，優(yōu)化Shuffle階段的性能和效率是MapReduce應(yīng)用程序優(yōu)化的一個關(guān)鍵方向。

3.3.6 排序的解釋

MapTask和ReduceTask對key進行排序是為了方便后續(xù)的數(shù)據(jù)處理和計算。

具體來說，對于MapTask而言，對輸出的key進行排序可以將具有相同key值的記錄聚合在一起，方便ReduceTask進行處理。

而對于ReduceTask而言，對輸入的key進行排序可以讓具有相同key值的記錄相鄰排列，方便進行聚合和計算。

一般來說，在Map任務(wù)中，對鍵值對進行快速排序的次數(shù)是一次，即將數(shù)據(jù)寫入環(huán)形緩沖區(qū)之前對其中的鍵值對進行排序。這是因為，對于同一個Map任務(wù)的輸出，在Map輸出的環(huán)形緩沖區(qū)中進行快速排序即可滿足Reduce任務(wù)在Shuffle階段的需求，而不需要進行額外的排序。

在Shuffle階段，如果存在多個環(huán)形緩沖區(qū)需要合并，Reduce任務(wù)會對它們進行歸并排序。這是因為，不同Map任務(wù)的輸出在Shuffle階段需要合并，而這些輸出之間的順序是無序的，因此需要進行排序以便進行合并。這次排序是對整個數(shù)據(jù)集進行的，而不是對單個Map任務(wù)的輸出進行的。

當Reduce任務(wù)接收到來自多個Map任務(wù)的中間結(jié)果時，它會對同一個分區(qū)內(nèi)的所有鍵值對進行排序。這里采用的排序算法一般也是歸并排序，因為歸并排序的時間復雜度是O(nlogn)，且適合對大量數(shù)據(jù)進行排序。這次排序也是對整個數(shù)據(jù)集進行的，而不是對單個Map任務(wù)的輸出進行的。

因此，總體來說，Shuffle階段需要進行多次排序，具體排序的次數(shù)可能因具體實現(xiàn)而有所不同。但無論是哪種具體實現(xiàn)，Shuffle階段都需要對整個數(shù)據(jù)集進行排序以便后續(xù)的計算和處理。

4 數(shù)據(jù)壓縮

4.1 壓縮算法對比

4.2 壓縮位置選擇

到此這篇關(guān)于Java大數(shù)據(jù)開發(fā)Hadoop MapReduce的文章就介紹到這了,更多相關(guān)Java Hadoop MapReduce內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評論