2019獨角獸企業重金招聘Python工程師標準>>>

1、計數器簡介

在許多情況下，一個用戶需要了解待分析的數據，盡管這并非所要執行的分析任務 的核心內容。以統計數據集中無效記錄數目的任務為例，如果發現無效記錄的比例 相當高，那么就需要認真思考為何存在如此多無效記錄。是所采用的檢測程序存在 缺陷，還是數據集質量確實很低，包含大量無效記錄？如果確定是數據集的質量問 題，則可能需要擴大數據集的規模，以增大有效記錄的比例，從而進行有意義的 分析。
計數器是一種收集作業統計信息的有效手段，用于質量控制或應用級統計。計數器 還可輔助診斷系統故障。如果需要將日志信息傳輸到map或reduce任務，更好的 方法通常是嘗試傳輸計數器值以監測某一特定事件是否發生。對于大型分布式作業 而言，使用計數器更為方便。首先，獲取計數器值比輸出日志更方便，其次，根據 計數器值統計特定事件的發生次數要比分析一堆日志文件容易得多。

2、內置計數器

Hadoop為每個作業維護若干內置計數器, 以描述該作業的各項指標。例如，某些計數器記錄已處理的字節數和記錄數，使用戶可監控已處理的輸入數據量和已產生的輸出數據量，并以此對 job 做適當的優化。

14/06/08 15:13:35 INFO mapreduce.Job: Counters: 46File System CountersFILE: Number of bytes read=159FILE: Number of bytes written=159447FILE: Number of read operations=0FILE: Number of large read operations=0FILE: Number of write operations=0HDFS: Number of bytes read=198HDFS: Number of bytes written=35HDFS: Number of read operations=6HDFS: Number of large read operations=0HDFS: Number of write operations=2Job Counters Launched map tasks=1Launched reduce tasks=1Rack-local map tasks=1Total time spent by all maps in occupied slots (ms)=3896Total time spent by all reduces in occupied slots (ms)=9006Map-Reduce FrameworkMap input records=3Map output records=12Map output bytes=129Map output materialized bytes=159Input split bytes=117Combine input records=0Combine output records=0Reduce input groups=4Reduce shuffle bytes=159Reduce input records=12Reduce output records=4Spilled Records=24Shuffled Maps =1Failed Shuffles=0Merged Map outputs=1GC time elapsed (ms)=13CPU time spent (ms)=3830Physical memory (bytes) snapshot=537718784Virtual memory (bytes) snapshot=7365263360Total committed heap usage (bytes)=2022309888Shuffle ErrorsBAD_ID=0CONNECTION=0IO_ERROR=0WRONG_LENGTH=0WRONG_MAP=0WRONG_REDUCE=0File Input Format Counters Bytes Read=81File Output Format Counters Bytes Written=35 計數器由其關聯任務維護，并定期傳到tasktracker，再由tasktracker傳給 jobtracker.因此，計數器能夠被全局地聚集。詳見第 hadoop 權威指南第170頁的“進度和狀態的更新”小節。與其他計數器（包括用戶定義的計數器）不同，內置的作業計數器實際上 由jobtracker維護，不必在整個網絡中發送。
一個任務的計數器值每次都是完整傳輸的，而非自上次傳輸之后再繼續數未完成的傳輸，以避免由于消息丟失而引發的錯誤。另外，如果一個任務在作業執行期間失 敗，則相關計數器值會減小。僅當一個作業執行成功之后，計數器的值才是完整可 靠的。

3、用戶定義的Java計數器

MapReduce允許用戶編寫程序來定義計數器，計數器的值可在mapper或reducer 中增加。多個計數器由一個Java枚舉(enum)類型來定義，以便對計數器分組。一 個作業可以定義的枚舉類型數量不限，各個枚舉類型所包含的字段數量也不限。枚 舉類型的名稱即為組的名稱，枚舉類型的字段就是計數器名稱。計數器是全局的。 換言之，MapReduce框架將跨所有map和reduce聚集這些計數器，并在作業結束 時產生一個最終結果。

Notice1：需要說明的是，不同的 hadoop 版本定義的方式會有些許差異。

（1）在0.20.x版本中使用counter很簡單,直接定義即可，如無此counter，hadoop會自動添加此counter.

Counter ct = context.getCounter("INPUT_WORDS", "count"); ct.increment(1); （2）在0.19.x版本中,需要定義enum
enum MyCounter {INPUT_WORDS }; reporter.incrCounter(MyCounter.INPUT_WORDS, 1); RunningJob job = JobClient.runJob(conf); Counters c = job.getCounters(); long cnt = c.getCounter(MyCounter.INPUT_WORDS);

Notice2：使用計數器需要清楚的是它們都存儲在jobTracker的內存里。Mapper/Reducer 任務序列化它們，連同更新狀態被發送。為了運行正常且jobTracker不會出問題，計數器的數量應該在10-100個，計數器不僅僅只用來聚合MapReduce job的統計值。新版本的hadoop限制了計數器的數量，以防給jobTracker帶來損害。你最不想看到的事情就是由于定義上百個計數器而使jobTracker宕機。
下面咱們來看一個計數器的實例（以下代碼請運行在 0.20.1 版本以上）：

3.1 測試數據：

hello world 2013 mapreduce hello world 2013 mapreduce hello world 2013 mapreduce

3.2 代碼：

/*** Project Name:CDHJobs* File Name:MapredCounter.java* Package Name:tmp* Date:2014-6-8下午2:12:48* Copyright (c) 2014, decli#qq.com All Rights Reserved.**/package tmp;import java.io.IOException; import java.util.StringTokenizer;import org.apache.commons.lang3.StringUtils; 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.Counter; import org.apache.hadoop.mapreduce.CounterGroup; import org.apache.hadoop.mapreduce.Counters; 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 WordCountWithCounter {static enum WordsNature {STARTS_WITH_DIGIT, STARTS_WITH_LETTER, ALL}/*** The map class of WordCount.*/public static class TokenCounterMapper 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);}}}/*** The reducer class of WordCount*/public static class TokenCounterReducer extends Reducer<Text, IntWritable, Text, IntWritable> {public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException,InterruptedException {int sum = 0;String token = key.toString();if (StringUtils.isNumeric(token)) {context.getCounter(WordsNature.STARTS_WITH_DIGIT).increment(1);} else if (StringUtils.isAlpha(token)) {context.getCounter(WordsNature.STARTS_WITH_LETTER).increment(1);}context.getCounter(WordsNature.ALL).increment(1);for (IntWritable value : values) {sum += value.get();}context.write(key, new IntWritable(sum));}}/*** The main entry point.*/public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job job = new Job(conf, "WordCountWithCounter");job.setJarByClass(WordCountWithCounter.class);job.setMapperClass(TokenCounterMapper.class);job.setReducerClass(TokenCounterReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);FileInputFormat.addInputPath(job, new Path("/tmp/dsap/rawdata/june/a.txt"));FileOutputFormat.setOutputPath(job, new Path("/tmp/dsap/rawdata/june/a_result"));int exitCode = job.waitForCompletion(true) ? 0 : 1;Counters counters = job.getCounters();Counter c1 = counters.findCounter(WordsNature.STARTS_WITH_DIGIT);System.out.println("-------------->>>>: " + c1.getDisplayName() + ": " + c1.getValue());// The below example shows how to get built-in counter groups that Hadoop provides basically.for (CounterGroup group : counters) {System.out.println("==========================================================");System.out.println("* Counter Group: " + group.getDisplayName() + " (" + group.getName() + ")");System.out.println(" number of counters in this group: " + group.size());for (Counter counter : group) {System.out.println(" ++++ " + counter.getDisplayName() + ": " + counter.getName() + ": "+ counter.getValue());}}System.exit(exitCode);} }

3.3 結果與計數器詳解

運行結果下面會一并給出。Counter有"組group"的概念，用于表示邏輯上相同范圍的所有數值。MapReduce job提供的默認Counter分為7個組，下面逐一介紹。這里也拿上面的測試數據來做詳細比對，我將會針對具體的計數器，挑選一些主要的簡述一下。?

... 前面省略 job 運行信息 xx 字 ...ALL=4STARTS_WITH_DIGIT=1STARTS_WITH_LETTER=3 -------------->>>>: STARTS_WITH_DIGIT: 1 ========================================================== #MapReduce job執行所依賴的數據來自于不同的文件系統，這個group表示job與文件系統交互的讀寫統計 * Counter Group: File System Counters (org.apache.hadoop.mapreduce.FileSystemCounter)number of counters in this group: 10#job讀取本地文件系統的文件字節數。假定我們當前map的輸入數據都來自于HDFS，那么在map階段，這個數據應該是0。但reduce在執行前，它 的輸入數據是經過shuffle的merge后存儲在reduce端本地磁盤中，所以這個數據就是所有reduce的總輸入字節數。++++ FILE: Number of bytes read: FILE_BYTES_READ: 159#map的中間結果都會spill到本地磁盤中，在map執行完后，形成最終的spill文件。所以map端這里的數據就表示map task往本地磁盤中總共寫了多少字節。與map端相對應的是，reduce端在shuffle時，會不斷地拉取map端的中間結果，然后做merge并 不斷spill到自己的本地磁盤中。最終形成一個單獨文件，這個文件就是reduce的輸入文件。 ++++ FILE: Number of bytes written: FILE_BYTES_WRITTEN: 159447++++ FILE: Number of read operations: FILE_READ_OPS: 0++++ FILE: Number of large read operations: FILE_LARGE_READ_OPS: 0++++ FILE: Number of write operations: FILE_WRITE_OPS: 0# 整個job執行過程中，只有map端運行時，才從HDFS讀取數據，這些數據不限于源文件內容，還包括所有map的split元數據。所以這個值應該比FileInputFormatCounters.BYTES_READ 要略大些。 ++++ HDFS: Number of bytes read: HDFS_BYTES_READ: 198#Reduce的最終結果都會寫入HDFS，就是一個job執行結果的總量。 ++++ HDFS: Number of bytes written: HDFS_BYTES_WRITTEN: 35++++ HDFS: Number of read operations: HDFS_READ_OPS: 6++++ HDFS: Number of large read operations: HDFS_LARGE_READ_OPS: 0++++ HDFS: Number of write operations: HDFS_WRITE_OPS: 2 ========================================================== #這個group描述與job調度相關的統計 * Counter Group: Job Counters (org.apache.hadoop.mapreduce.JobCounter)number of counters in this group: 5#Job在被調度時，如果啟動了一個data-local(源文件的幅本在執行map task的taskTracker本地) ++++ Data-local map tasks #當前job為某些map task的執行保留了slot，總共保留的時間是多少 ++++ FALLOW_SLOTS_MILLIS_MAPS/REDUCES#所有map task占用slot的總時間，包含執行時間和創建/銷毀子JVM的時間++++ SLOTS_MILLIS_MAPS/REDUCES# 此job啟動了多少個map task ++++ Launched map tasks: TOTAL_LAUNCHED_MAPS: 1# 此job啟動了多少個reduce task ++++ Launched reduce tasks: TOTAL_LAUNCHED_REDUCES: 1++++ Rack-local map tasks: RACK_LOCAL_MAPS: 1++++ Total time spent by all maps in occupied slots (ms): SLOTS_MILLIS_MAPS: 3896++++ Total time spent by all reduces in occupied slots (ms): SLOTS_MILLIS_REDUCES: 9006 ========================================================== #這個Counter group包含了相當多地job執行細節數據。這里需要有個概念認識是：一般情況下，record就表示一行數據，而相對地byte表示這行數據的大小是 多少，這里的group表示經過reduce merge后像這樣的輸入形式{"aaa", [5, 8, 2, …]}。 * Counter Group: Map-Reduce Framework (org.apache.hadoop.mapreduce.TaskCounter)number of counters in this group: 20#所有map task從HDFS讀取的文件總行數 ++++ Map input records: MAP_INPUT_RECORDS: 3#map task的直接輸出record是多少，就是在map方法中調用context.write的次數，也就是未經過Combine時的原生輸出條數 ++++ Map output records: MAP_OUTPUT_RECORDS: 12# Map的輸出結果key/value都會被序列化到內存緩沖區中，所以這里的bytes指序列化后的最終字節之和 ++++ Map output bytes: MAP_OUTPUT_BYTES: 129++++ Map output materialized bytes: MAP_OUTPUT_MATERIALIZED_BYTES: 159# #與map task 的split相關的數據都會保存于HDFS中，而在保存時元數據也相應地存儲著數據是以怎樣的壓縮方式放入的，它的具體類型是什么，這些額外的數據是 MapReduce框架加入的，與job無關，這里記錄的大小就是表示額外信息的字節大小++++ Input split bytes: SPLIT_RAW_BYTES: 117#Combiner是為了減少盡量減少需要拉取和移動的數據，所以combine輸入條數與map的輸出條數是一致的。++++ Combine input records: COMBINE_INPUT_RECORDS: 0# 經過Combiner后，相同key的數據經過壓縮，在map端自己解決了很多重復數據，表示最終在map端中間文件中的所有條目數 ++++ Combine output records: COMBINE_OUTPUT_RECORDS: 0#Reduce總共讀取了多少個這樣的groups ++++ Reduce input groups: REDUCE_INPUT_GROUPS: 4#Reduce端的copy線程總共從map端抓取了多少的中間數據，表示各個map task最終的中間文件總和 ++++ Reduce shuffle bytes: REDUCE_SHUFFLE_BYTES: 159#如果有Combiner的話，那么這里的數值就等于map端Combiner運算后的最后條數，如果沒有，那么就應該等于map的輸出條數 ++++ Reduce input records: REDUCE_INPUT_RECORDS: 12#所有reduce執行后輸出的總條目數 ++++ Reduce output records: REDUCE_OUTPUT_RECORDS: 4#spill過程在map和reduce端都會發生，這里統計在總共從內存往磁盤中spill了多少條數據 ++++ Spilled Records: SPILLED_RECORDS: 24#每個reduce幾乎都得從所有map端拉取數據，每個copy線程拉取成功一個map的數據，那么增1，所以它的總數基本等于 reduce number * map number ++++ Shuffled Maps : SHUFFLED_MAPS: 1# copy線程在抓取map端中間數據時，如果因為網絡連接異?；蚴荌O異常，所引起的shuffle錯誤次數 ++++ Failed Shuffles: FAILED_SHUFFLE: 0#記錄著shuffle過程中總共經歷了多少次merge動作 ++++ Merged Map outputs: MERGED_MAP_OUTPUTS: 1#通過JMX獲取到執行map與reduce的子JVM總共的GC時間消耗 ++++ GC time elapsed (ms): GC_TIME_MILLIS: 13++++ CPU time spent (ms): CPU_MILLISECONDS: 3830++++ Physical memory (bytes) snapshot: PHYSICAL_MEMORY_BYTES: 537718784++++ Virtual memory (bytes) snapshot: VIRTUAL_MEMORY_BYTES: 7365263360++++ Total committed heap usage (bytes): COMMITTED_HEAP_BYTES: 2022309888 ========================================================== #這組內描述Shuffle過程中的各種錯誤情況發生次數，基本定位于Shuffle階段copy線程抓取map端中間數據時的各種錯誤。 * Counter Group: Shuffle Errors (Shuffle Errors)number of counters in this group: 6#每個map都有一個ID，如attempt_201109020150_0254_m_000000_0，如果reduce的copy線程抓取過來的元數據中這個ID不是標準格式，那么此Counter增加 ++++ BAD_ID: BAD_ID: 0#表示copy線程建立到map端的連接有誤 ++++ CONNECTION: CONNECTION: 0#Reduce的copy線程如果在抓取map端數據時出現IOException，那么這個值相應增加 ++++ IO_ERROR: IO_ERROR: 0#map端的那個中間結果是有壓縮好的有格式數據，所有它有兩個length信息：源數據大小與壓縮后數據大小。如果這兩個length信息傳輸的有誤(負值)，那么此Counter增加++++ WRONG_LENGTH: WRONG_LENGTH: 0#每個copy線程當然是有目的:為某個reduce抓取某些map的中間結果，如果當前抓取的map數據不是copy線程之前定義好的map，那么就表示把數據拉錯了++++ WRONG_MAP: WRONG_MAP: 0#與上面描述一致，如果抓取的數據表示它不是為此reduce而準備的，那還是拉錯數據了。 ++++ WRONG_REDUCE: WRONG_REDUCE: 0 ========================================================== #這個group表示map task讀取文件內容(總輸入數據)的統計 * Counter Group: File Input Format Counters (org.apache.hadoop.mapreduce.lib.input.FileInputFormatCounter)number of counters in this group: 1 # Map task的所有輸入數據(字節)，等于各個map task的map方法傳入的所有value值字節之和。 ++++ Bytes Read: BYTES_READ: 81 ========================================================== ##這個group表示reduce task輸出文件內容(總輸出數據)的統計 * Counter Group: File Output Format Counters (org.apache.hadoop.mapreduce.lib.output.FileOutputFormatCounter)number of counters in this group: 1++++ Bytes Written: BYTES_WRITTEN: 35 ========================================================== # 自定義計數器的統計 * Counter Group: tmp.WordCountWithCounter$WordsNature (tmp.WordCountWithCounter$WordsNature)number of counters in this group: 3++++ ALL: ALL: 4++++ STARTS_WITH_DIGIT: STARTS_WITH_DIGIT: 1++++ STARTS_WITH_LETTER: STARTS_WITH_LETTER: 3

4、最后的問題：

如果想要在 MapReduce 中實現一個類似計數器的“全局變量”，可以在 map、reduce 中以任意數據類型、任意修改變量值，并在 main 函數中回調獲取該怎么辦呢？

5、Refer:

（1）An Example of Hadoop MapReduce Counter

http://diveintodata.org/2011/03/15/an-example-of-hadoop-mapreduce-counter/

（2）Hadoop Tutorial Series, Issue #3: Counters In Action

http://www.philippeadjiman.com/blog/2010/01/07/hadoop-tutorial-series-issue-3-counters-in-action/

（3）Controlling Hadoop MapReduce Job recursion

http://codingwiththomas.blogspot.com/2011/04/controlling-hadoop-job-recursion.html

（4）MapReduce Design Patterns（chapter 2 （part 3））（四）

http://blog.csdn.net/cuirong1986/article/details/8456923

（5）[hadoop源碼閱讀][5]-counter的使用和默認counter的含義

http://www.cnblogs.com/xuxm2007/archive/2012/06/15/2551030.html

轉載于:https://my.oschina.net/leejun2005/blog/276891

總結

以上是生活随笔為你收集整理的MapReduce 计数器简介的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。