2. 程式人生 > >Hadoop系列之-MapReduce

Hadoop系列之-MapReduce

阿新 • • 發佈：2018-12-29

Hadoop系列之-MapReduce

      MapReduce在Hadoop1.x中直接執行在HDFS上，由JobTracker和TaskTracker負責排程。在Hadoop2.x中執行在YARN上。面對大量資料的離線計算，MR無非是很好的選擇，但如果需要高及時性的大資料計算，Spark、Storm是更合適的選擇。MapReduce的核心設計理念是分散式計算，簡單說就是移動計算而不移動資料。

  • 基本功能模組

    YARN : 負責資源管理和排程

    MapReduceApplicationMaster : 負責切分、任務排程、任務監控和容錯等

    MapTask/ReduceTask : 任務驅動引擎

  • 每個MapReduce作業對應一個MapReduceApplicationMaster，MRAppMaster負責整體任務排程，YARN將資源分配給MRAppMaster，MRAppMaster再將任務分配給MapTask/ReduceTask

  • MapReduceApplicationMaster容錯

    失敗後，由YARN重新啟動

    任務失敗後，MRAppMaster重新申請資源

概念

  • MapReduce

    • MR整體框架

      MR的整個過程由以下構成：

      1. Split

        將資料按配置好的大小（如一個block是64M將其切分為3份，也就是25M+25M+14M）切分，並傳給不同的MapTask。

      2. Map

        Map對檔案逐行讀出並解析，解析完畢後以K,V的格式輸出。

      3. Shuffle

        在交付ReduceTask之前，輸出的過程中可以配置Shuffle，Shuffle可對資料先Sort再Combiner，以減小Reduce的工作量。在ReduceTask端，Shuffle會對資料先進行聚合，相同Key的所有Value會被整合為一個以Key為首的Value陣列

      4. Reduce

        ReduceTask可根據不同需求對shuffle後的Value陣列進行整合統計。如果一次MR計算不夠，可執行多次計算。

      在這裡插入圖片描述

    • 假設有一個檔案儲存這Deer Bear River等word，我們要計算這個檔案中各word的總數，那麼MR的整套過程是這樣的

      1. 將檔案按大小Split為多份

      2. Map階段按行讀入，按K，V的結果集輸出

      3. Shuffle對Map後的資料進行Sort、聚合

      4. Reduce對Shuffle後的結果進行統計並輸出

      在這裡插入圖片描述

  • Shuffler

          每個MapTask都有一個記憶體緩衝區（預設是100M），儲存著map的輸出結果，當緩衝區快滿的時候需要將緩衝區的資料以一個臨時檔案的方式存放到磁碟，溢寫是由單獨執行緒完成，不影響往緩衝區寫map結果的執行緒（split.percent，預設是0.8）。當溢寫執行緒啟動後，需要對這80M空間內的key做排序（Sort。）

          假如client設定過Combiner，那麼現在就是使用Combinerd的時候了。將由相同key的key/value對的value加起來，減少溢寫到磁碟的資料量。

          當整個map task結束後在對磁碟中這個map task產生的所有臨時檔案做合併（Merge），對於“word1”就是這樣的：{word1:[5,8,2,…]}，假如有Combiner，{word1[15]}，最終產生一個檔案。

          reduce從tasktracker copy資料，copy過來的資料會先放入記憶體緩衝區中，這裡的緩衝區大小要比map端的更為靈活，它基於JVM的heap size設定。merge有三種形式：記憶體到記憶體，記憶體到磁碟，磁碟到磁碟。merge從不同的task tracker上拿到的資料。

    在這裡插入圖片描述

  • Combiner

          每一個map可能會產生大量的輸出。combiner的作用就是在map端對輸出先做一次合併，以減少傳輸到reducer的資料量。combiner最基本是實現本地key的歸併，combiner具有類似本地的reduce功能。如果不用combiner，那麼，所有的結果都由reduce完成，效率會相對地下，如果使用combiner，先完成的map會在本地聚合，提升速度。

    在這裡插入圖片描述

配置

配置筆者就不在此贅述，有需要可以去讀官網 http://hadoop.apache.org/docs/

MR執行時Hadoop各程序之間的互動

  1. 客戶端提交一個mr的jar包給JobClient(提交方式：hadoop jar <jar在unix的url> <main方法所在的類的全類名> <引數>)

  2. JobClient通過RPC和JobTracker進行通訊，返回一個存放jar包的地址（HDFS）和jobId

  3. client將jar包寫入到HDFS當中(path = hdfs上的地址 + jobId)

  4. 開始提交任務(任務的描述資訊，不是jar, 包括jobid，jar存放的位置，配置資訊等等)

  5. JobTracker進行初始化任務

  6. 讀取HDFS上的要處理的檔案，開始計算輸入分片，每一個分片對應一個MapperTask

  7. TaskTracker通過心跳機制領取任務（任務的描述資訊）

  8. 下載所需的jar，配置檔案等

  9. TaskTracker啟動一個java child子程序，用來執行具體的任務（MapperTask或ReducerTask）

  10. 將結果寫入到HDFS當中

小試牛刀

      我們做一個簡單的MapReduce，現在有三個檔案a.txt、b.txt、c.txt，每個檔案中有大量單詞，通過MR計算後，我們要得出的結果是：wordname a.txt->28 b.txt->11 c.txt->22，並將其儲存在一個檔案中

a.txt

name age sex flower
name flower sex age
name flower sex age
name flower sex age
name flower sex age

b.txt

name        sex age
name flower sex age
name flower     age
     flower sex age
name flower sex

c.txt

name flower sex age
name flower sex age
name flower sex age

pom.xml

<dependencies>

		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-common</artifactId>
			<version>2.5.2</version>
			<exclusions>
				<exclusion>
					<artifactId>jdk.tools</artifactId>
					<groupId>jdk.tools</groupId>
				</exclusion>
			</exclusions>
		</dependency>
		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-hdfs</artifactId>
			<version>2.5.2</version>
		</dependency>

		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-mapreduce-client-core</artifactId>
			<version>2.5.2</version>
			<exclusions>
				<exclusion>
					<artifactId>jdk.tools</artifactId>
					<groupId>jdk.tools</groupId>
				</exclusion>
			</exclusions>
		</dependency>

		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-mapreduce-client-common</artifactId>
			<version>2.5.2</version>
		</dependency>

		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-mapreduce-client-jobclient</artifactId>
			<version>2.5.2</version>
		</dependency>

		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-yarn-common</artifactId>
			<version>2.5.2</version>
		</dependency>

	</dependencies>

Java程式碼


import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * map+reducce
 * 實現倒序索引(計算各個單詞在不同檔案中的個數)
 * 輸出結果: word a.txt->28 b.txt->11 c.txt->22
 * 
 * 將 a.txt、b.txt、c.txt傳到 hdfs://host:port/words
 * 執行 hadoop jar ./NormalInverseIndex.jar cn.hadoop.Inverse hdfs://host:port/words hdfs://host:port/result
 * 
 */
public class NormalInverseIndex {

  private static Text k = new Text();
  private static Text v = new Text();

  // 
  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
	
    Job job = Job.getInstance(conf);
    // 設定jar
    job.setJarByClass(NormalInverseIndex.class);

    // 設定Mapper相關的屬性
    job.setMapperClass(IndexMapper.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(Text.class);
    FileInputFormat.setInputPaths(job, new Path(args[0]));

    // 設定Reducer相關屬性
    job.setReducerClass(IndexReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(Text.class);
    FileOutputFormat.setOutputPath(job, new Path(args[1]));

//    job.setCombinerClass(IndexCombiner.class);

    // 提交任務
    job.waitForCompletion(true);
  }
  
  /**
   * output：
   *    key：word
   *    value：fileName
   */
  public static class IndexMapper extends Mapper<LongWritable, Text, Text, Text> {

    @Override
    protected void map(LongWritable key, Text value,
        Mapper<LongWritable, Text, Text, Text>.Context context)
        throws IOException, InterruptedException {
      StringTokenizer st = new StringTokenizer(value.toString());
      String fileName = ((FileSplit) context.getInputSplit()).getPath().getName();
      while (st.hasMoreTokens()) {
        k.set(st.nextToken());
        v.set(fileName);
        context.write(k, v);
      }
    }
  }

  /**
   * output：
   *    key：word
   *    value：a.txt->28 b.txt->11 c.txt->22
   */
  public static class IndexReducer extends Reducer<Text, Text, Text, Text> {

    @Override
    protected void reduce(Text key, Iterable<Text> values,
        Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
      Map<String, Integer> fileNameMap = new HashMap<String, Integer>();
      for (Text fileName : values) {
        if (fileNameMap.get(fileName.toString()) == null) {
          fileNameMap.put(fileName.toString(), 1);
        } else {
          fileNameMap.put(fileName.toString(), fileNameMap.get(fileName.toString())+1);
        }
      }
      k.set(key.toString());
      String vVal = "";
      Set<Entry<String, Integer>> set = fileNameMap.entrySet();
      for (Iterator<Entry<String, Integer>> iterator = set.iterator(); iterator.hasNext();) {
        Map.Entry<String, Integer> en = iterator.next();
        vVal += " "+en.getKey()+"->"+en.getValue().toString();
      }
      v.set(vVal.substring(1));
      context.write(k, v);
    }
  }

}

 
 
 
 
 
 
 
 
 
 
 

相關推薦

Hadoop系列之-MapReduce

Hadoop系列之-MapReduce       MapReduce在Hadoop1.x中直接執行在HDFS上，由JobTracker和TaskTracker負責排程。在Hadoop2.x中執行在YARN上。面對大量資料的離線

Hadoop系列－－Hadoop核心之MapReduce的原理

1 MapReduce核心原理 　　“分而治之，平行計算”是MapReduce的核心原理，其實也是大資料處理的中心思想。 1.1 分而治之 　　在MapReduce中，分而治之，就是， 　　 一個任務分成多個小的子任務(map)，並行執行後，

Hadoop系列之四：MapReduce進階

1、mapper和reducer MapReduce對資料的處理分為兩個階段：map階段和reduce階段，這兩個階段分別由使用者開發的map函式和reduce函式完成，在MapReduce執行環境中執行時，它們也分別被稱為mapper和reducer。鍵值對(key-v

hadoop系列三:mapreduce的使用

count 明顯 blank api park size 當前 java mapreduce 轉載請在頁首明顯處註明作者與出處 一：說明 此為大數據系列的一些博文，有空的話會陸續更新，包含大數據的一些內容，如hadoop,spark,storm,機器學習等。

大資料Hadoop系列之Hadoop服務開機自啟動配置

1.  編寫執行指令碼 $ sudo cd /etc/init.d $ sudo vi hadoop #!/bin/bash #chkconfig:35 95 1 #description:script to start/stop hadoop su - hadoop

MapReduce系列之MapReduce的輸出

針對前面介紹的輸入格式，MapReduce也有相應的輸出格式。預設情況下只有一個 Reduce，輸出只有一個檔案，預設檔名為 part-r-00000，輸出檔案的個數與 Reduce 的個數一致。 如果有兩個Reduce，輸出結果就有兩個檔案，第一個為part-r-00000，第二個為part-r-

MapReduce系列之MapReduce的輸入

檔案是 MapReduce 任務資料的初始儲存地。正常情況下，輸入檔案一般是儲存在 HDFS 裡面。這些檔案的格式可以是任意的：我們可以使用基於行的日誌檔案， 也可以使用二進位制格式，多行輸入記錄或者其它一些格式。這些檔案一般會很大，達到數十GB，甚至更大。那麼 MapReduce 是如何讀取這些資

MapReduce系列之MapReduce任務處理流程

MapReduce處理資料的流程一般是這樣的： 1、從HDFS上讀取資料，因為是分散式與平行計算，需要將資料劃分給多個MapReduce任務。HDFS儲存檔案也是分塊的，每個MapReduce的輸入一般是和HDFS的資料塊是對應的。也就是說一個HDFS資料塊作為一個MapReduce任務的

hadoop系列之linux系統模板的製作

CentOS7_64 位作業系統模板搭建 說明：鑑於平時使用虛擬機器做各種測試的頻率非常高，難免有很多重複工作。這裡以 CentOS-7-x86_64-Minimal-1804 安裝為基礎安裝了虛擬機器，然後再做了如下配置與安裝。將該虛擬機器做成模板，可以直接拷

Hadoop系列之hadoop環境搭建本地模式

  1.1 Hadoop環境搭建 Hadoop 的環境可以有多種方式，比如本地模式（standalone）、偽分散式、完全分散式以及 HA 模式。參考：    1.1.1Hadoop安裝包   ​​​​​​​核心配置檔案 Hadoop 主要有四個核心配置檔案，

hadoop系列之偽分散式環境搭建及測試驗證

Hadoop2.x 偽分散式環境搭建及測試驗證 作者：Dennis 日期：2018-08-09 前置條件： Linux 虛擬機器一臺，版本為 CentOS 7.4，假設 IP 地址為 192.168.159.181，並修改如下： 修改/etc/hostname 的

Hadoop系列之完全分散式環境搭建及測試驗證

前置條件： 已製作好CentOS 虛擬機器模板。 該模板安裝好Oracle JDK，且 JAVA_HOME 值為/usr/java/jdk1.8.0_171-amd64/jre。 準備三臺虛擬機器 搭建完全 Hadoop 分散式最少需要三臺伺服器，假設三臺伺服器

hadoop系列之yarn

YARN YARN概述 YARN，即 Yet Another Resource Negotiator 的縮寫， 它是 Hadoop 資源管理系統，是在 Hadoop2 以後引入的。它在整個 Hadoop 中的位置如下： YARN 與 MapR

大資料Hadoop系列之Hadoop Web控制檯新增身份驗證

1. 背景介紹 本文件介紹如何配置Hadoop HTTP Web控制檯以要求使用者身份驗證。 預設情況下，Hadoop HTTP Web控制檯（ResourceManager，NameNode，NodeManagers和DataNodes）允許訪問而無需任何形式的身份驗證

Hadoop系列之-Hbase資料設計問題

Hadoop系列之-Hbase資料設計問題       Hbase是一個基於HDFS，可通過MapReduce計算的分散式列式資料庫，每行資料都有一個rowKey作為當前資料的唯一Sign，當然，也可通過Column Famil

Hadoop系列之-HDFS

HDFS       HDFS是Hadoop整體架構的底層儲存系統，從資料結構上來說，它適合儲存半結構化、非結構化、多維的資料，如果實時性要求不高，那麼它也可儲存關係性很強資料的資料。從資料量來說，它的分散式體系和容錯機制可容納

Hadoop系列之（一）CentOS7安裝配置及SSH無密碼驗證配置

由於最近在進行關於Hadoop的學習，開始重新搭建一套Hadoop的環境，和之前自己搭建的環境的區別在於作業系統的版本，本次採用的為CentOS7.2的版本，7相對於6的版本改動較大。一些命令都不太一樣，其中在其核心中加了Docker，因此在後期裝相關軟體時不需要重新安

大資料與Hadoop系列之壓縮(一)

Hadoop壓縮簡介Hadoop作為一個較通用的海量數椐處理平臺，在使用壓縮方式方面，主要考慮壓縮速度和壓縮檔案的可分割性.所有的壓縮演算法都會考慮時間和空間的權衡，更快的壓縮和解壓縮速度通常會耗費更多的交間（壓縮比較低）例如：通過gzip命令壓縮資料時，使用者可以設定不同的

hadoop系列之HDFS 原理與實戰

HDFS檔案系統 HDFS簡介 HDFS 是 Hadoop Distributed File System 的簡稱，即 Hadoop 分散式檔案系統。它起源於谷歌發表的 GFS 論文， 是該論文的開源實現，也是整個大資料的基礎。 HDFS 專門為解決大資料的儲存問題而產生

Hadoop系列之OutputCollector

該介面的程式碼如下 public interface OutputCollector<K, V> { void collect(K var1, V var2) throws IO