Hadoop 新 MapReduce 框架 Yarn 詳解

時間  2019-11-24
標籤 hadoop mapreduce 框架 yarn 詳解 欄目 Hadoop 简体版
原文   原文鏈接

Hadoop MapReduceV2(Yarn) 框架簡介java

原 Hadoop MapReduce 框架的問題node

對於業界的大數據存儲及分佈式處理系統來講，Hadoop 是耳熟能詳的卓越開源分佈式文件存儲及處理框架，對於 Hadoop 框架的介紹在此再也不累述，讀者可參考 Hadoop 官方簡介。使用和學習過老 Hadoop 框架（0.20.0 及以前版本）的同仁應該很熟悉以下的原 MapReduce 框架圖：linux


圖 1.Hadoop 原 MapReduce 架構
圖 1.Hadoop 原 MapReduce 架構
web

從上圖中能夠清楚的看出原 MapReduce 程序的流程及設計思路：shell

  1. 首先用戶程序 (JobClient) 提交了一個 job，job 的信息會發送到 Job Tracker 中，Job Tracker 是 Map-reduce 框架的中心，他須要與集羣中的機器定時通訊 (heartbeat), 須要管理哪些程序應該跑在哪些機器上，須要管理全部 job 失敗、重啓等操做。apache

  2. TaskTracker 是 Map-reduce 集羣中每臺機器都有的一個部分，他作的事情主要是監視本身所在機器的資源狀況。編程

  3. TaskTracker 同時監視當前機器的 tasks 運行情況。TaskTracker 須要把這些信息經過 heartbeat 發送給 JobTracker，JobTracker 會蒐集這些信息以給新提交的 job 分配運行在哪些機器上。上圖虛線箭頭就是表示消息的發送 - 接收的過程。安全

能夠看得出原來的 map-reduce 架構是簡單明瞭的，在最初推出的幾年，也獲得了衆多的成功案例，得到業界普遍的支持和確定，但隨着分佈式系統集羣的規模和其工做負荷的增加，原框架的問題逐漸浮出水面，主要的問題集中以下：bash

  1. JobTracker 是 Map-reduce 的集中處理點，存在單點故障。服務器

  2. JobTracker 完成了太多的任務，形成了過多的資源消耗，當 map-reduce job 很是多的時候，會形成很大的內存開銷，潛在來講，也增長了 JobTracker fail 的風險，這也是業界廣泛總結出老 Hadoop 的 Map-Reduce 只能支持 4000 節點主機的上限。

  3. 在 TaskTracker 端，以 map/reduce task 的數目做爲資源的表示過於簡單，沒有考慮到 cpu/ 內存的佔用狀況，若是兩個大內存消耗的 task 被調度到了一塊，很容易出現 OOM。

  4. 在 TaskTracker 端，把資源強制劃分爲 map task slot 和 reduce task slot, 若是當系統中只有 map task 或者只有 reduce task 的時候，會形成資源的浪費，也就是前面提過的集羣資源利用的問題。

  5. 源代碼層面分析的時候，會發現代碼很是的難讀，經常由於一個 class 作了太多的事情，代碼量達 3000 多行，，形成 class 的任務不清晰，增長 bug 修復和版本維護的難度。

  6. 從操做的角度來看，如今的 Hadoop MapReduce 框架在有任何重要的或者不重要的變化 ( 例如 bug 修復，性能提高和特性化 ) 時，都會強制進行系統級別的升級更新。更糟的是，它無論用戶的喜愛，強制讓分佈式集羣系統的每個用戶端同時更新。這些更新會讓用戶爲了驗證他們以前的應用程序是否是適用新的 Hadoop 版本而浪費大量時間。

新 Hadoop Yarn 框架原理及運做機制

從業界使用分佈式系統的變化趨勢和 hadoop 框架的長遠發展來看，MapReduce 的 JobTracker/TaskTracker 機制須要大規模的調整來修復它在可擴展性，內存消耗，線程模型，可靠性和性能上的缺陷。在過去的幾年中，hadoop 開發團隊作了一些 bug 的修復，可是最近這些修復的成本愈來愈高，這代表對原框架作出改變的難度愈來愈大。

爲從根本上解決舊 MapReduce 框架的性能瓶頸，促進 Hadoop 框架的更長遠發展，從 0.23.0 版本開始，Hadoop 的 MapReduce 框架徹底重構，發生了根本的變化。新的 Hadoop MapReduce 框架命名爲 MapReduceV2 或者叫 Yarn，其架構圖以下圖所示：


圖 2. 新的 Hadoop MapReduce 框架（Yarn）架構
圖 2. 新的 Hadoop MapReduce 框架（Yarn）架構

重構根本的思想是將 JobTracker 兩個主要的功能分離成單獨的組件，這兩個功能是資源管理和任務調度 / 監控。新的資源管理器全局管理全部應用程序計算資源的分配，每個應用的 ApplicationMaster 負責相應的調度和協調。一個應用程序無非是一個單獨的傳統的 MapReduce 任務或者是一個 DAG( 有向無環圖 ) 任務。ResourceManager 和每一臺機器的節點管理服務器可以管理用戶在那臺機器上的進程並能對計算進行組織。

事實上，每個應用的 ApplicationMaster 是一個詳細的框架庫，它結合從 ResourceManager 得到的資源和 NodeManager 協同工做來運行和監控任務。

上圖中 ResourceManager 支持分層級的應用隊列，這些隊列享有集羣必定比例的資源。從某種意義上講它就是一個純粹的調度器，它在執行過程當中不對應用進行監控和狀態跟蹤。一樣，它也不能重啓因應用失敗或者硬件錯誤而運行失敗的任務。

ResourceManager 是基於應用程序對資源的需求進行調度的 ; 每個應用程序須要不一樣類型的資源所以就須要不一樣的容器。資源包括：內存，CPU，磁盤，網絡等等。能夠看出，這同現 Mapreduce 固定類型的資源使用模型有顯著區別，它給集羣的使用帶來負面的影響。資源管理器提供一個調度策略的插件，它負責將集羣資源分配給多個隊列和應用程序。調度插件能夠基於現有的能力調度和公平調度模型。

上圖中 NodeManager 是每一臺機器框架的代理，是執行應用程序的容器，監控應用程序的資源使用狀況 (CPU，內存，硬盤，網絡 ) 而且向調度器彙報。

每個應用的 ApplicationMaster 的職責有：向調度器索要適當的資源容器，運行任務，跟蹤應用程序的狀態和監控它們的進程，處理任務的失敗緣由。

新舊 Hadoop MapReduce 框架比對

讓咱們來對新舊 MapReduce 框架作詳細的分析和對比，能夠看到有如下幾點顯著變化：

首先客戶端不變，其調用 API 及接口大部分保持兼容，這也是爲了對開發使用者透明化，使其沒必要對原有代碼作大的改變 ( 詳見 2.3 Demo 代碼開發及詳解)，可是原框架中核心的 JobTracker 和 TaskTracker 不見了，取而代之的是 ResourceManager, ApplicationMaster 與 NodeManager 三個部分。

咱們來詳細解釋這三個部分，首先 ResourceManager 是一箇中心的服務，它作的事情是調度、啓動每個 Job 所屬的 ApplicationMaster、另外監控 ApplicationMaster 的存在狀況。細心的讀者會發現：Job 裏面所在的 task 的監控、重啓等等內容不見了。這就是 AppMst 存在的緣由。ResourceManager 負責做業與資源的調度。接收 JobSubmitter 提交的做業，按照做業的上下文 (Context) 信息，以及從 NodeManager 收集來的狀態信息，啓動調度過程，分配一個 Container 做爲 App Mstr

NodeManager 功能比較專注，就是負責 Container 狀態的維護，並向 RM 保持心跳。

ApplicationMaster 負責一個 Job 生命週期內的全部工做，相似老的框架中 JobTracker。但注意每個 Job（不是每一種）都有一個 ApplicationMaster，它能夠運行在 ResourceManager 之外的機器上。

Yarn 框架相對於老的 MapReduce 框架什麼優點呢？咱們能夠看到：

  1. 這個設計大大減少了 JobTracker（也就是如今的 ResourceManager）的資源消耗，而且讓監測每個 Job 子任務 (tasks) 狀態的程序分佈式化了，更安全、更優美。

  2. 在新的 Yarn 中，ApplicationMaster 是一個可變動的部分，用戶能夠對不一樣的編程模型寫本身的 AppMst，讓更多類型的編程模型可以跑在 Hadoop 集羣中，能夠參考 hadoop Yarn 官方配置模板中的 mapred-site.xml 配置。

  3. 對於資源的表示之內存爲單位 ( 在目前版本的 Yarn 中，沒有考慮 cpu 的佔用 )，比以前以剩餘 slot 數目更合理。

  4. 老的框架中，JobTracker 一個很大的負擔就是監控 job 下的 tasks 的運行情況，如今，這個部分就扔給 ApplicationMaster 作了，而 ResourceManager 中有一個模塊叫作 ApplicationsMasters( 注意不是 ApplicationMaster)，它是監測 ApplicationMaster 的運行情況，若是出問題，會將其在其餘機器上重啓。

  5. Container 是 Yarn 爲了未來做資源隔離而提出的一個框架。這一點應該借鑑了 Mesos 的工做，目前是一個框架，僅僅提供 java 虛擬機內存的隔離 ,hadoop 團隊的設計思路應該後續能支持更多的資源調度和控制 , 既然資源表示成內存量，那就沒有了以前的 map slot/reduce slot 分開形成集羣資源閒置的尷尬狀況。

新的 Yarn 框架相對舊 MapRduce 框架而言，其配置文件 , 啓停腳本及全局變量等也發生了一些變化，主要的改變以下：


表 1. 新舊 Hadoop 腳本 / 變量 / 位置變化表

改變項 原框架中 新框架中（Yarn） 備註
配置文件位置 ${hadoop_home_dir}/conf ${hadoop_home_dir}/etc/hadoop/ Yarn 框架也兼容老的 ${hadoop_home_dir}/conf 位置配置，啓動時會檢測是否存在老的 conf 目錄，若是存在將加載 conf 目錄下的配置，不然加載 etc 下配置
啓停腳本 ${hadoop_home_dir}/bin/start（stop）-all.sh ${hadoop_home_dir}/sbin/start（stop）-dfs.sh
${hadoop_home_dir}/bin/start(stop)-all.sh		 新的 Yarn 框架中啓動分佈式文件系統和啓動 Yarn 分離，啓動 / 中止分佈式文件系統的命令位於 ${hadoop_home_dir}/sbin 目錄下，啓動 / 中止 Yarn 框架位於 ${hadoop_home_dir}/bin/ 目錄下
JAVA_HOME 全局變量 ${hadoop_home_dir}/bin/start-all.sh 中 ${hadoop_home_dir}/etc/hadoop/hadoop-env.sh
${hadoop_home_dir}/etc/hadoop/Yarn-env.sh		 Yarn 框架中因爲啓動 hdfs 分佈式文件系統和啓動 MapReduce 框架分離，JAVA_HOME 須要在 hadoop-env.sh 和 Yarn-env.sh 中分別配置
HADOOP_LOG_DIR 全局變量 不須要配置 ${hadoop_home_dir}/etc/hadoop/hadoop-env.sh 老框架在 LOG，conf，tmp 目錄等均默認爲腳本啓動的當前目錄下的 log,conf，tmp 子目錄
Yarn 新框架中 Log 默認建立在 Hadoop 用戶的 home 目錄下的 log 子目錄，所以最好在 ${hadoop_home_dir}/etc/hadoop/hadoop-env.sh 配置 HADOOP_LOG_DIR，不然有可能會由於你啓動 hadoop 的用戶的 .bashrc 或者 .bash_profile 中指定了其餘的 PATH 變量而形成日誌位置混亂，而該位置沒有訪問權限的話啓動過程當中會報錯


因爲新的 Yarn 框架與原 Hadoop MapReduce 框架相比變化較大，核心的配置文件中不少項在新框架中已經廢棄，而新框架中新增了不少其餘配置項，看下錶所示會更加清晰：


表 2. 新舊 Hadoop 框架配置項變化表

配置文件 配置項 Hadoop 0.20.X 配置 Hadoop 0.23.X 配置 說明
core-site.xml 系統默認分佈式文件 URI fs.default.name fs.defaultFS
hdfs-site.xml DFS name node 存放 name table 的目錄 dfs.name.dir dfs.namenode.name.dir 新框架中 name node 分紅 dfs.namenode.name.dir( 存放 naname table 和 dfs.namenode.edits.dir（存放 edit 文件），默認是同一個目錄
DFS data node 存放數據 block 的目錄 dfs.data.dir dfs.datanode.data.dir 新框架中 DataNode 增長更多細節配置，位於 dfs.datanode. 配置項下，如 dfs.datanode.data.dir.perm（datanode local 目錄默認權限）；dfs.datanode.address（datanode 節點監聽端口）；等
分佈式文件系統數據塊複製數 dfs.replication dfs.replication 新框架與老框架一致，值建議配置爲與分佈式 cluster 中實際的 DataNode 主機數一致
mapred-site.xml Job 監控地址及端口 mapred.job.tracker 無 新框架中已改成 Yarn-site.xml 中的 resouceManager 及 nodeManager 具體配置項，新框架中歷史 job 的查詢已從 Job tracker 剝離，納入單獨的 mapreduce.jobtracker.jobhistory 相關配置，
第三方 MapReduce 框架 無 mapreduce.framework.name 新框架支持第三方 MapReduce 開發框架以支持如 SmartTalk/DGSG 等非 Yarn 架構，注意一般狀況下這個配置的值都設置爲 Yarn，若是沒有配置這項，那麼提交的 Yarn job 只會運行在 locale 模式，而不是分佈式模式。
Yarn-site.xml The address of the applications manager interface in the RM 無 Yarn.resourcemanager.address 新框架中 NodeManager 與 RM 通訊的接口地址
The address of the scheduler interface 無 Yarn.resourcemanager.scheduler.address 同上，NodeManger 須要知道 RM 主機的 scheduler 調度服務接口地址
The address of the RM web application 無 Yarn.resourcemanager.webapp.address 新框架中各個 task 的資源調度及運行情況經過經過該 web 界面訪問
The address of the resource tracker interface 無 Yarn.resourcemanager.resource-tracker.address 新框架中 NodeManager 須要向 RM 報告任務運行狀態供 Resouce 跟蹤，所以 NodeManager 節點主機須要知道 RM 主機的 tracker 接口地址


回頁首

Hadoop Yarn 框架 Demo 示例

Demo 場景介紹：Weblogic 應用服務器日誌分析

瞭解了 hadoop 新的 Yarn 框架的架構和思路後，咱們用一個 Demo 示例來檢驗新 Yarn 框架下 Map-Reduce 程序的開發部署。

咱們考慮以下應用場景：用戶的生產系統由多臺 Weblogic 應用服務器組成，天天須要每臺對應用服務器的日誌內容進行檢查，統計其日誌級別和日誌模塊的總數。

WebLogic 的日誌範例以下圖所示：


圖 3.Weblogic 日誌示例
圖 3.Weblogic 日誌示例

如上圖所示，<Info> 爲 weblogic 的日誌級別，<Security>，<Management> 爲 Weblogic 的日誌模塊，咱們主要分析 loglevel 和 logmodule 這兩個維度分別在 WebLogic 日誌中出現的次數，天天須要統計出 loglevel 和 logmodule 分別出現的次數總數。

Demo 測試環境 Yarn 框架搭建

因爲 Weblogic 應用服務器分佈於不一樣的主機，且日誌數據量巨大，咱們採用 hadoop 框架將 WebLogic 各個應用服務器主機上創建分佈式目錄，天天將 WebLogic 日誌裝載進 hadoop 分佈式文件系統，而且編寫基於 Yarn 框架的 MapReduce 程序對日誌進行處理，分別統計出 LogLevel 和 Logmodule 在日誌中出現的次數並計算總量，而後輸出到分佈式文件系統中，輸出目錄命名精確到小時爲後綴以便區分每次 Demo 程序運行的處理結果。

咱們搭建一個 Demo 測試環境以驗證 Yarn 框架下分佈式程序處理該案例的功能，以兩臺虛擬機做爲該 Demo 的運行平臺，兩機均爲 Linux 操做系統，機器 hostname 爲 OEL 和 Stephen，OEL 做爲 NameNode 和 ResouceManager 節點主機，64 位，Stephen 做爲 DataNode 和 NodeManager 節點主機，32 位（Hadoop 支持異構性）, 具體以下：


表 3.Demo 測試環境表

主機名 角色 備註
OEL(192.168.137.8) NameNode 節點主機
ResourceManager 主機		 linux 操做系統
32bit
Stephen(192.168.l37.2) DataNode 節點主機
NodeManager 主機		 linux 操做系統
64bit


咱們把 hadoop 安裝在兩臺測試機的 /hadoop 文件系統目錄下，安裝後的 hadoop 根目錄爲：/hadoop/hadoop-0.23.0，規劃分佈式文件系統存放於 /hadoop/dfs 的本地目錄，對應分佈式系統中的目錄爲 /user/oracle/dfs

咱們根據 Yarn 框架要求，分別在 core-site.xml 中配置分佈式文件系統的 URL，詳細以下：


清單 1.core-site.xml 配置

 <configuration> 
   <property> 
   <name>fs.defaultFS</name> 
   <value>hdfs://192.168.137.8:9100</value> 
   </property> 
 </configuration>


在 hdfs-site.xml 中配置 nameNode，dataNode 的本地目錄信息，詳細以下：


清單 2.hdfs-site.xml 配置

 <configuration> 
 <property> 
  <name>dfs.namenode.name.dir</name> 
  <value>/hadoop/dfs/name</value> 
  <description>  </description> 
 </property> 

 <property> 
  <name>dfs.datanode.data.dir</name> 
  <value>/hadoop/dfs/data</value> 
  <description> </description> 
 </property> 

 <property> 
   <name>dfs.replication</name> 
   <value>2</value> 
 </property> 

 </configuration>


在 mapred-site.xml 中配置其使用 Yarn 框架執行 map-reduce 處理程序，詳細以下：


清單 3.mapred-site.xml 配置

 <configuration> 
  <property> 
  <name>mapreduce.framework.name</name> 
  <value>Yarn</value> 
  </property> 
 </configuration>


最後在 Yarn-site.xml 中配置 ResourceManager，NodeManager 的通訊端口，web 監控端口等，詳細以下：


清單 4.Yarn-site.xml 配置

 <?xml version="1.0"?> 
 <configuration> 

 <!-- Site specific YARN configuration properties --> 
  <property> 
  <name>Yarn.nodemanager.aux-services</name> 
  <value>mapreduce.shuffle</value> 
  </property> 
  <property> 
  <description>The address of the applications manager interface in the RM.</description> 
  <name>Yarn.resourcemanager.address</name> 
  <value>192.168.137.8:18040</value> 
  </property> 

  <property> 
  <description>The address of the scheduler interface.</description> 
  <name>Yarn.resourcemanager.scheduler.address</name> 
  <value>192.168.137.8:18030</value> 
  </property> 

  <property> 
  <description>The address of the RM web application.</description> 
  <name>Yarn.resourcemanager.webapp.address</name> 
  <value>192.168.137.8:18088</value> 
  </property> 
  
  <property> 
  <description>The address of the resource tracker interface.</description> 
  <name>Yarn.resourcemanager.resource-tracker.address</name> 
  <value>192.168.137.8:8025</value> 
  </property> 
 </configuration>


具體配置項的含義，在 hadoop 官方網站有詳細的說明，讀者能夠參見 hadoop 0.23.0 官方配置模板。

Demo 代碼開發及詳解

如下咱們詳細介紹一下新的 Yarn 框架下針對該應用場景的 Demo 代碼的開發， 在 Demo 程序的每一個類都有詳細的註釋和說明，Yarn 開發爲了兼容老版本，API 變化不大，能夠參考 官方 Hadoop Yarn 框架 API。

在 Map 程序中，咱們以行號爲 key，行文本爲 value 讀取每一行 WebLogic 日誌輸入，將 loglevel 和 logmodule 的值讀出做爲 Map 處理後的新的 key 值，因爲一行中 loglevel 和 logmodule 的出現次數應該惟一，因此經 Map 程序處理後的新的 record 記錄的 value 應該都爲 1：


清單 5. Map 業務邏輯

 public static class MapClass extends Mapper<Object, Text, Text, IntWritable> 
  { 
  private Text record = new Text(); 
  private static final IntWritable recbytes = new IntWritable(1); 
  public void map(Object key, Text value,Context context) 
    throws IOException,InterruptedException {  
  String line = value.toString(); 
 // 沒有配置 RecordReader，因此默認採用 line 的實現，
 //key 就是行號，value 就是行內容，
 // 按行 key-value 存放每行 loglevel 和 logmodule 內容
  if (line == null || line.equals("")) 
    return; 
  String[] words = line.split("> <"); 
  if (words == null || words.length < 2) 
    return; 
  String logLevel = words[1]; 
  String moduleName = words[2]; 
  
  record.clear(); 
  record.set(new StringBuffer("logLevel::").append(logLevel).toString()); 
  context.write(record, recbytes); 
  // 輸出日誌級別統計結果，經過 logLevel:: 做爲前綴來標示。
  
  record.clear(); 
   record.set(new StringBuffer("moduleName::").append(moduleName).toString()); 
  context.write(record, recbytes); 
   // 輸出模塊名的統計結果，經過 moduleName:: 做爲前綴來標示
  }  
  }


因爲有 loglevel 和 logmodule 兩部分的分析工做，咱們設定兩個 Reduce 來分別處理這兩部分，loglevel 的交給 reduce1，logmodule 交給 reduce2。所以咱們編寫 Patitioner 類，根據 Map 傳過來的 Key 中包含的 logLevel 和 moduleName 的前綴，來分配到不一樣的 Reduce：


清單 6.Partition 業務邏輯

public static class PartitionerClass extends Partitioner<Text, IntWritable>
 {
 public int getPartition(Text key, IntWritable value, int numPartitions)
 {
 if (numPartitions >= 2)//Reduce 個數，判斷 loglevel 仍是 logmodule 的統計，分配到不一樣的 Reduce
   if (key.toString().startsWith("logLevel::"))
   return 0;
   else if(key.toString().startsWith("moduleName::"))
   return 1;
   else return 0;
 else
   return 0;
 }

 }


在 Reduce 程序中，累加併合並 loglevel 和 logmodule 的出現次數


清單 7. Reduce 業務邏輯

 public static class ReduceClass 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 tmp = 0; 
                for (IntWritable val : values) { 
                    tmp = tmp + val.get();                                     
                } 
                result.set(tmp); 
                context.write(key, result);// 輸出最後的彙總結果
            }    
        }


以上完成了 MapReduce 的主要處理邏輯，對於程序入口，咱們使用 Hadoop 提供的 Tools 工具包方便的進行 May-Reduce 程序的啓動和 Map/Reduce 對應處理 class 的配置。


清單 8. Main 執行類

 import java.io.File; 
 import java.io.IOException; 
 import java.text.SimpleDateFormat; 
 import java.util.Date; 
 import java.util.Iterator; 
 import org.apache.hadoop.conf.Configuration; 
 import org.apache.hadoop.conf.Configured; 
 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.Reducer; 
 import org.apache.hadoop.mapreduce.Mapper; 
 import org.apache.hadoop.mapreduce.Partitioner; 
 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
 import org.apache.hadoop.util.Tool; 
 import org.apache.hadoop.util.ToolRunner; 
 public class LogAnalysiser extends Configured implements Tool { 
   public static void main(String[] args) 
  { 
    try 
  { 
  int res; 
  res = ToolRunner.run(new Configuration(),new LogAnalysiser(), args); 
  System.exit(res); 
  } catch (Exception e) 
  { 
  e.printStackTrace(); 
  } 
  } 
  public int run(String[] args) throws Exception 
  { 
  if (args == null || args.length <2) 
  { 
  System.out.println("need inputpath and outputpath"); 
  return 1; 
  } 
  String inputpath = args[0]; 
  String outputpath = args[1]; 
  String shortin = args[0]; 
  String shortout = args[1]; 
  if (shortin.indexOf(File.separator) >= 0) 
  shortin = shortin.substring(shortin.lastIndexOf(File.separator)); 
  if (shortout.indexOf(File.separator) >= 0) 
  shortout = shortout.substring(shortout.lastIndexOf(File.separator)); 
  SimpleDateFormat formater = new SimpleDateFormat("yyyy.MM.dd.HH.mm"); 
  shortout = new StringBuffer(shortout).append("-") 
  .append(formater.format(new Date())).toString(); 
  
  
  if (!shortin.startsWith("/")) 
  shortin = "/" + shortin; 
  if (!shortout.startsWith("/")) 
  shortout = "/" + shortout; 
  shortin = "/user/oracle/dfs/" + shortin; 
  shortout = "/user/oracle/dfs/" + shortout;   
  File inputdir = new File(inputpath); 
  File outputdir = new File(outputpath); 
  
  if (!inputdir.exists() || !inputdir.isDirectory()) 
  { 
  System.out.println("inputpath not exist or isn't dir!"); 
  return 0; 
  } 
  if (!outputdir.exists()) 
  { 
  new File(outputpath).mkdirs(); 
  } 
 // 如下注釋的是 hadoop 0.20.X 老版本的 Job 代碼，在 hadoop0.23.X 新框架中已經大大簡化
 //   Configuration conf = getConf(); 
 //   JobConf job = new JobConf(conf, LogAnalysiser.class);   
 //    JobConf conf = new JobConf(getConf(),LogAnalysiser.class);// 構建 Config 
 //    conf.setJarByClass(MapClass.class); 
 //    conf.setJarByClass(ReduceClass.class); 
 //    conf.setJarByClass(PartitionerClass.class); 
 //    conf.setJar("hadoopTest.jar"); 
 //    job.setJar("hadoopTest.jar"); 

 // 如下是新的 hadoop 0.23.X Yarn 的 Job 代碼

 job job = new Job(new Configuration()); 
    job.setJarByClass(LogAnalysiser.class); 
    job.setJobName("analysisjob"); 
    job.setOutputKeyClass(Text.class);// 輸出的 key 類型，在 OutputFormat 會檢查
    job.setOutputValueClass(IntWritable.class); // 輸出的 value 類型，在 OutputFormat 會檢查
    job.setJarByClass(LogAnalysiser.class); 
    job.setMapperClass(MapClass.class); 
    job.setCombinerClass(ReduceClass.class); 
    job.setReducerClass(ReduceClass.class); 
    job.setPartitionerClass(PartitionerClass.class); 
    job.setNumReduceTasks(2);// 強制須要有兩個 Reduce 來分別處理流量和次數的統計
    FileInputFormat.setInputPaths(job, new Path(shortin));//hdfs 中的輸入路徑
    FileOutputFormat.setOutputPath(job,new Path(shortout));//hdfs 中輸出路徑
    
    Date startTime = new Date(); 
    System.out.println("Job started: " + startTime); 
    job.waitForCompletion(true);    
    Date end_time = new Date(); 
    System.out.println("Job ended: " + end_time); 
    System.out.println("The job took " + 
    (end_time.getTime() - startTime.getTime()) /1000 + " seconds."); 
    // 刪除輸入和輸出的臨時文件
 //    fileSys.copyToLocalFile(new Path(shortout),new Path(outputpath)); 
 //    fileSys.delete(new Path(shortin),true); 
 //    fileSys.delete(new Path(shortout),true); 
    return 0; 
  } 
 }


Demo 部署及運行

Demo 輸入輸出的控制

本 demo 中咱們將從 Weblogic 日誌目錄中拷貝原始待處理日誌文件做爲 Yarn 程序的輸入，使用 hadoop dfs 命令將其放入分佈式目錄的 input 目錄，處理完後將生成以時間戳爲文件目錄後綴的輸出目錄

Weblogic 日誌存放的原始目錄位於：/u01/app/Oracle/Middleware/user_projects/domains/test_domain/AdminServer/logs

分佈式文件系統中的輸入目錄：/user/oracle/dfs/input

分佈式文件系統中的輸出目錄：/user/oracle/dfs/output_%YYYY-MM-DD-hh-mm%

Demo 打包和部署

可使用 JDeveloper 或者 Eclipse 等 IDE 工具將開發的 Hadoop Demo 代碼打包爲 jar，並指定 Main 類爲 LoyAnalyze，本文中咱們採用 JDeveloper 打包 Demo 代碼，以下圖示例：


圖 4.Yarn Demo 程序打包示例
圖 4.Yarn Demo 程序打包示例

Demo 執行與跟蹤

咱們在 OEL 主機（NameNode&ResourceManager 主機，192.168.137.8）上啓動 dfs 分佈式文件系統：


圖 5. 啓動 Demo dfs 文件系統
圖 5. 啓動 Demo dfs 文件系統

從上圖能夠看出 dfs 分佈式文件系統已經在 OEL 和 Stephen 主機上成功啓動，咱們經過默認的分佈式文件系統 Web 監控 端口 http://192.168.137.8:50070（也能夠在上文中 core-site.xml 中配置 dfs.namenode.http-address 項指定其餘端口 ) 來驗證其文件系統狀況：


圖 6.hadoop 文件系統 web 監控頁面
圖 6.hadoop 文件系統 web 監控頁面

從上圖中咱們能夠看到 /user/oracle/dfs 分佈式文件系統已成功創建。

接下來咱們在 NameNode 主機（OEL，192.168.137.8）上啓動 Yarn 框架：


圖 7. 啓動 Demo Yarn 框架
圖 7. 啓動 Demo Yarn 框架

從上圖咱們能夠看到 ResouceManager 在 OEL 主機上成功啓動，NodeManager 進程在 Stephen 節點主機上也已經啓動，至此整個新的 Hadoop Yarn 框架已經成功啓動。

咱們將打好的 testHadoop.jar 包上傳至 NameNode 主機（OEL）的 /hadoop/hadoop-0.23.0/ 根目錄下，咱們使用 Hadoop 自帶的 hadoop 命令行工具執行 Demo 的 jar 包，具體步驟爲，先使用 hadoop dfs 命令將輸入文件（weblogic 原始日誌）拷貝至 dfs 分佈式目錄的 input 輸入目錄，清理 dfs 分佈式目錄下的 output 輸出子目錄。而後使用 hadoop jar 命令執行 testHadoop 的 jar 包。

執行 Demo 的 shell 腳本示例以下：

 ./bin/hadoop dfs -rmr /user/oracle/dfs/output* 
 ./bin/hadoop dfs -rmr /user/oracle/dfs/input 
 ./bin/hadoop dfs -mkdir /user/oracle/dfs/input 
 ./bin/hadoop dfs -copyFromLocal ./input/*.log /user/oracle/dfs/input/ 
 ./bin/hadoop jar ./hadoopTest.jar /hadoop/hadoop-0.23.0/input 
                 /hadoop/hadoop-0.23.0/output


清單 9.Demo 執行腳本

而後咱們使用上文中的腳本啓動 demo 並執行：


圖 8.Demo 程序運行
圖 8.Demo 程序運行

查看大圖

從上圖的 console 輸出中咱們能夠看到 Demo 程序的結果和各項統計信息輸出，下面咱們經過 Web 監控界面詳細中觀察程序執行的執行流程和步驟細節。

Job 啓動後咱們能夠經過 ResourceManager 的 Web 端口（在上文中 Yarn-site.xml 配置文件中 Yarn.resourcemanager.webapp.address 配置項） http://192.168.137.8:18088 來監控其 job 的資源調度。


圖 9. 接收請求和生成 job application
圖 9. 接收請求和生成 job application

查看大圖

上圖中咱們能夠看到 Yarn 框架接受到客戶端請求 , 如上圖所示 ID 爲 application_1346564668712_0003 的 job 已是 accepted 狀態

咱們點擊該 ID 的連接進入到該 application 的 Map-Reduce 處理監控頁面，該界面中有動態分配的 ApplicationMaster 的 Web 跟蹤端口能夠監視 MapReduce 程序的步驟細節


圖 10.hadoop MapReduce Application Web 監控頁面 (1)
圖 10.hadoop MapReduce Application Web 監控頁面 (1)

點擊上圖中 ApplicationMaster 的 URL 能夠進入該 ApplicationMaster 負責管理的 Job 的具體 Map-Reduce 運行狀態：


圖 11.hadoop MasterApplication Web 監控頁面（2）
圖 11.hadoop MasterApplication Web 監控頁面（2）

上圖中咱們能夠看到 ID 爲 application_1346564668712_0003 的 Job 正在執行，有 2 個 Map 進程，已經處理完畢，有 2 個 Reduce 正在處理，這跟咱們程序設計預期的是同樣的。

當狀態變爲 successful 後，進入 dfs 文件系統能夠看到，輸出的 dfs 文件系統已經生成，位置位於 /user/oracle/dfs 下，目錄名爲 output-2012.09.02.13.52，能夠看到格式和命名方式與 Demo 設計是一致的，以下圖所示：


圖 12.Demo 輸出目錄（1）
圖 12.Demo 輸出目錄（1）

咱們進入具體的輸出目錄，能夠清楚的看到程序處理的輸出結果，正如咱們 Demo 中設計的，兩個 Reduce 分別生成了兩個輸出文件，分別是 part-r-00000 和 part-r-00001，對應 Module 和 Log Level 的處理輸出信息：


圖 13.Demo 輸出目錄（2）
圖 13.Demo 輸出目錄（2）

點擊 part-r-00000 的輸出文件連接，能夠看到程序處理後的 log level 的統計信息：


圖 14.Demo 輸出結果（1）
圖 14.Demo 輸出結果（1）

點擊 part-r-00001 的輸出文件連接，能夠看到程序處理後 Module 的統計信息：


圖 15.Demo 輸出結果（2）
圖 15.Demo 輸出結果（2）

至此咱們基於新的 Yarn 框架的 Demo 徹底成功運行，實現功能與預期設計徹底一致，運行狀態和 NameNode/DataNode 部署，Job/MapReduece 程序的調度均和設計一致。讀者可參考該 Demo 的配置及代碼進行修改，作爲實際生產環境部署和實施的基礎。

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程