Hadoop學習之路（十五）MapReduce的多Job串聯和全局計數器

MapReduce 多 Job 串聯

需求

一個稍復雜點的處理邏輯往往需要多個 MapReduce 程序串聯處理，多 job 的串聯可以借助 MapReduce 框架的 JobControl 實現

實例

以下有兩個 MapReduce 任務，分別是 Flow 的 SumMR 和 SortMR，其中有依賴關系：SumMR 的輸出是 SortMR 的輸入，所以 SortMR 的啟動得在 SumMR 完成之后

Configuration conf1 = new Configuration();
        Configuration conf2 = new Configuration();
        
        Job job1 = Job.getInstance(conf1);
        Job job2 = Job.getInstance(conf2);
            
        job1.setJarByClass(MRScore3.class);
        job1.setMapperClass(MRMapper3_1.class);
        //job.setReducerClass(ScoreReducer3.class);
        
        
        job1.setMapOutputKeyClass(IntWritable.class);
        job1.setMapOutputValueClass(StudentBean.class);
        job1.setOutputKeyClass(IntWritable.class);
        job1.setOutputValueClass(StudentBean.class);
        
        job1.setPartitionerClass(CoursePartitioner2.class);
        
        job1.setNumReduceTasks(4);
        
        Path inputPath = new Path("D:\\MR\\hw\\work3\\input");
        Path outputPath = new Path("D:\\MR\\hw\\work3\\output_hw3_1");
            
        FileInputFormat.setInputPaths(job1, inputPath);
        FileOutputFormat.setOutputPath(job1, outputPath);
        
        job2.setMapperClass(MRMapper3_2.class);
        job2.setReducerClass(MRReducer3_2.class);
        
        job2.setMapOutputKeyClass(IntWritable.class);
        job2.setMapOutputValueClass(StudentBean.class);
        job2.setOutputKeyClass(StudentBean.class);
        job2.setOutputValueClass(NullWritable.class);
        
        Path inputPath2 = new Path("D:\\MR\\hw\\work3\\output_hw3_1");
        Path outputPath2 = new Path("D:\\MR\\hw\\work3\\output_hw3_end");
            
        FileInputFormat.setInputPaths(job2, inputPath2);
        FileOutputFormat.setOutputPath(job2, outputPath2);
        
        JobControl control = new JobControl("Score3"); ControlledJob aJob = new ControlledJob(job1.getConfiguration()); ControlledJob bJob = new ControlledJob(job2.getConfiguration());
        // 設置作業依賴關系
        bJob.addDependingJob(aJob); control.addJob(aJob); control.addJob(bJob); Thread thread = new Thread(control); thread.start(); while(!control.allFinished()) { thread.sleep(1000); } System.exit(0); 

MapReduce 全局計數器

MapReduce計數器是什么？

  計數器是用來記錄job的執行進度和狀態的。它的作用可以理解為日志。我們可以在程序的某個位置插入計數器，記錄數據或者進度的變化情況。

MapReduce計數器能做什么？

  MapReduce 計數器（Counter）為我們提供一個窗口，用於觀察 MapReduce Job 運行期的各種細節數據。對MapReduce性能調優很有幫助，MapReduce性能優化的評估大部分都是基於這些 Counter 的數值表現出來的。

MapReduce 都有哪些內置計數器？

  MapReduce 自帶了許多默認Counter，現在我們來分析這些默認 Counter 的含義，方便大家觀察 Job 結果，如輸入的字節數、輸出的字節數、Map端輸入/輸出的字節數和條數、Reduce端的輸入/輸出的字節數和條數等。下面我們只需了解這些內置計數器，知道計數器組名稱（groupName）和計數器名稱（counterName），以后使用計數器會查找groupName和counterName即可。

  1、任務計數器

    在任務執行過程中，任務計數器采集任務的相關信息，每個作業的所有任務的結果會被聚集起來。例如，MAP_INPUT_RECORDS 計數器統計每個map任務輸入記錄的總數，並在一個作業的所有map任務上進行聚集，使得最終數字是整個作業的所有輸入記錄的總數。任務計數器由其關聯任務維護，並定期發送給TaskTracker，再由TaskTracker發送給 JobTracker。因此，計數器能夠被全局地聚集。下面我們分別了解各種任務計數器。

　　1）MapReduce 任務計數器

      MapReduce 任務計數器的 groupName為org.apache.hadoop.mapreduce.TaskCounter，它包含的計數器如下表所示

計數器名稱	說明
map 輸入的記錄數（MAP_INPUT_RECORDS）	作業中所有 map 已處理的輸入記錄數。每次 RecorderReader 讀到一條記錄並將其傳給 map 的 map() 函數時，該計數器的值增加。
map 跳過的記錄數（MAP_SKIPPED_RECORDS）	作業中所有 map 跳過的輸入記錄數。
map 輸入的字節數（MAP_INPUT_BYTES）	作業中所有 map 已處理的未經壓縮的輸入數據的字節數。每次 RecorderReader 讀到一條記錄並 將其傳給 map 的 map() 函數時，該計數器的值增加
分片split的原始字節數（SPLIT_RAW_BYTES）	由 map 讀取的輸入-分片對象的字節數。這些對象描述分片元數據（文件的位移和長度），而不是分片的數據自身，因此總規模是小的
map 輸出的記錄數（MAP_OUTPUT_RECORDS）	作業中所有 map 產生的 map 輸出記錄數。每次某一個 map 的Context 調用 write() 方法時，該計數器的值增加
map 輸出的字節數（MAP_OUTPUT_BYTES）	作業中所有 map 產生的 未經壓縮的輸出數據的字節數。每次某一個 map 的 Context 調用 write() 方法時，該計數器的值增加。
map 輸出的物化字節數（MAP_OUTPUT_MATERIALIZED_BYTES）	map 輸出后確實寫到磁盤上的字節數；若 map 輸出壓縮功能被啟用，則會在計數器值上反映出來
combine 輸入的記錄數（COMBINE_INPUT_RECORDS）	作業中所有 Combiner（如果有）已處理的輸入記錄數。Combiner 的迭代器每次讀一個值，該計數器的值增加。
combine 輸出的記錄數（COMBINE_OUTPUT_RECORDS）	作業中所有 Combiner（如果有）已產生的輸出記錄數。每當一個 Combiner 的 Context 調用 write() 方法時，該計數器的值增加。
reduce 輸入的組（REDUCE_INPUT_GROUPS）	作業中所有 reducer 已經處理的不同的碼分組的個數。每當某一個 reducer 的 reduce() 被調用時，該計數器的值增加。
reduce 輸入的記錄數（REDUCE_INPUT_RECORDS）	作業中所有 reducer 已經處理的輸入記錄的個數。每當某個 reducer 的迭代器讀一個值時，該計數器的值增加。如果所有 reducer 已經處理完所有輸入， 則該計數器的值與計數器 “map 輸出的記錄” 的值相同
reduce 輸出的記錄數（REDUCE_OUTPUT_RECORDS）	作業中所有 map 已經產生的 reduce 輸出記錄數。每當某一個 reducer 的 Context 調用 write() 方法時，該計數器的值增加。
reduce 跳過的組數（REDUCE_SKIPPED_GROUPS）	作業中所有 reducer 已經跳過的不同的碼分組的個數。
reduce 跳過的記錄數（REDUCE_SKIPPED_RECORDS）	作業中所有 reducer 已經跳過輸入記錄數。
reduce 經過 shuffle 的字節數（REDUCE_SHUFFLE_BYTES）	shuffle 將 map 的輸出數據復制到 reducer 中的字節數。
溢出的記錄數（SPILLED_RECORDS）	作業中所有 map和reduce 任務溢出到磁盤的記錄數
CPU 毫秒（CPU_MILLISECONDS）	總計的 CPU 時間，以毫秒為單位，由/proc/cpuinfo獲取
物理內存字節數（PHYSICAL_MEMORY_BYTES）	一個任務所用物理內存的字節數，由/proc/cpuinfo獲取
虛擬內存字節數（VIRTUAL_MEMORY_BYTES）	一個任務所用虛擬內存的字節數，由/proc/cpuinfo獲取
有效的堆字節數（COMMITTED_HEAP_BYTES）	在 JVM 中的總有效內存量（以字節為單位），可由Runtime().getRuntime().totaoMemory()獲取。
GC 運行時間毫秒數（GC_TIME_MILLIS）	在任務執行過程中，垃圾收集器（garbage collection）花費的時間（以毫秒為單位）， 可由 GarbageCollector MXBean.getCollectionTime()獲取；該計數器並未出現在1.x版本中。
由 shuffle 傳輸的 map 輸出數（SHUFFLED_MAPS）	有 shuffle 傳輸到 reducer 的 map 輸出文件數。
失敗的 shuffle 數（SHUFFLE_MAPS）	在 shuffle 過程中，發生拷貝錯誤的 map 輸出文件數，該計數器並沒有包含在 1.x 版本中。
被合並的 map 輸出數	在 shuffle 過程中，在 reduce 端被合並的 map 輸出文件數，該計數器沒有包含在 1.x 版本中。

　　2）文件系統計數器

      文件系統計數器的 groupName為org.apache.hadoop.mapreduce.FileSystemCounter，它包含的計數器如下表所示

　　

計數器名稱	說明
文件系統的讀字節數（BYTES_READ）	由 map 和 reduce 等任務在各個文件系統中讀取的字節數，各個文件系統分別對應一個計數器，可以是 Local、HDFS、S3和KFS等。
文件系統的寫字節數（BYTES_WRITTEN）	由 map 和 reduce 等任務在各個文件系統中寫的字節數。

 　　3）FileInputFormat 計數器

      FileInputFormat 計數器的 groupName為org.apache.hadoop.mapreduce.lib.input.FileInputFormatCounter，它包含的計數器如下表所示，計數器名稱列的括號（）內容即為counterName

計數器名稱	說明
讀取的字節數（BYTES_READ）	由 map 任務通過 FileInputFormat 讀取的字節數。

 

　　4）FileOutputFormat 計數器

      FileOutputFormat 計數器的 groupName為org.apache.hadoop.mapreduce.lib.input.FileOutputFormatCounter，它包含的計數器如下表所示

計數器名稱	說明
寫的字節數（BYTES_WRITTEN）	由 map 任務（針對僅含 map 的作業）或者 reduce 任務通過 FileOutputFormat 寫的字節數。

 

2、作業計數器

    作業計數器由 JobTracker（或者 YARN）維護，因此無需在網絡間傳輸數據，這一點與包括 “用戶定義的計數器” 在內的其它計數器不同。這些計數器都是作業級別的統計量，其值不會隨着任務運行而改變。 作業計數器計數器的 groupName為org.apache.hadoop.mapreduce.JobCounter，它包含的計數器如下表所示

計數器名稱	說明
啟用的map任務數（TOTAL_LAUNCHED_MAPS）	啟動的map任務數，包括以“推測執行” 方式啟動的任務。
啟用的 reduce 任務數（TOTAL_LAUNCHED_REDUCES）	啟動的reduce任務數，包括以“推測執行” 方式啟動的任務。
失敗的map任務數（NUM_FAILED_MAPS）	失敗的map任務數。
失敗的 reduce 任務數（NUM_FAILED_REDUCES）	失敗的reduce任務數。
數據本地化的 map 任務數（DATA_LOCAL_MAPS）	與輸入數據在同一節點的 map 任務數。
機架本地化的 map 任務數（RACK_LOCAL_MAPS）	與輸入數據在同一機架范圍內、但不在同一節點上的 map 任務數。
其它本地化的 map 任務數（OTHER_LOCAL_MAPS）	與輸入數據不在同一機架范圍內的 map 任務數。由於機架之間的寬帶資源相對較少，Hadoop 會盡量讓 map 任務靠近輸入數據執行，因此該計數器值一般比較小。
map 任務的總運行時間（SLOTS_MILLIS_MAPS）	map 任務的總運行時間，單位毫秒。該計數器包括以推測執行方式啟動的任務。
reduce 任務的總運行時間（SLOTS_MILLIS_REDUCES）	reduce任務的總運行時間，單位毫秒。該值包括以推測執行方式啟動的任務。
在保留槽之后，map任務等待的總時間（FALLOW_SLOTS_MILLIS_MAPS）	在為 map 任務保留槽之后所花費的總等待時間，單位是毫秒。
在保留槽之后，reduce 任務等待的總時間（FALLOW_SLOTS_MILLIS_REDUCES）	在為 reduce 任務保留槽之后，花在等待上的總時間，單位為毫秒。

 

計數器的該如何使用？

  下面我們來介紹如何使用計數器。

  1、定義計數器

    1)枚舉聲明計數器

// 自定義枚舉變量Enum 
Counter counter = context.getCounter(Enum enum)

 2)自定義計數器

/ 自己命名groupName和counterName 
Counter counter = context.getCounter(String groupName,String counterName)

 2、為計數器賦值

    1)初始化計數器

counter.setValue(long value);// 設置初始值

　2)計數器自增

counter.increment(long incr);// 增加計數

 3、獲取計數器的值

　　1) 獲取枚舉計數器的值

Configuration conf = new Configuration(); 
Job job = new Job(conf, "MyCounter"); 
job.waitForCompletion(true); 
Counters counters=job.getCounters(); 
Counter counter=counters.findCounter(LOG_PROCESSOR_COUNTER.BAD_RECORDS_LONG);// 查找枚舉計數器，假如Enum的變量為BAD_RECORDS_LONG 
long value=counter.getValue();//獲取計數值

　　2) 獲取自定義計數器的值

Configuration conf = new Configuration(); 
Job job = new Job(conf, "MyCounter"); 
job.waitForCompletion(true); 
Counters counters = job.getCounters(); 
Counter counter=counters.findCounter("ErrorCounter","toolong");// 假如groupName為ErrorCounter，counterName為toolong 
long value = counter.getValue();// 獲取計數值

　　3) 獲取內置計數器的值

Configuration conf = new Configuration(); 
Job job = new Job(conf, "MyCounter"); 
job.waitForCompletion(true); 
Counters counters=job.getCounters(); 
// 查找作業運行啟動的reduce個數的計數器，groupName和counterName可以從內置計數器表格查詢（前面已經列舉有） 
Counter counter=counters.findCounter("org.apache.hadoop.mapreduce.JobCounter","TOTAL_LAUNCHED_REDUCES");// 假如groupName為org.apache.hadoop.mapreduce.JobCounter，counterName為TOTAL_LAUNCHED_REDUCES 
long value=counter.getValue();// 獲取計數值

　　4) 獲取所有計數器的值

Configuration conf = new Configuration(); 
Job job = new Job(conf, "MyCounter"); 
Counters counters = job.getCounters(); 
for (CounterGroup group : counters) { 
  for (Counter counter : group) { 
    System.out.println(counter.getDisplayName() + ": " + counter.getName() + ": "+ counter.getValue()); 
  } 
}