「counter」[MapReduce] Counter

counter

Definition

Counter

Counter是hadoop mapreduce framework所提供的一个工具，用来追踪mapreduce job的进度。

Counter表示全局的计数器，由MapReduce framework或者APPlication定义。每一个Counter有一个与之相关联的name和long类型的value。Counter通过Enum类来定义，Enum类定义一个CounterGroup, 每个field顶一个一个Counter的name.

CounterGroup

CounterGroup是由逻辑上属于一个group的所有的Counter所组成。一个CounterGroup对应一个枚举类，每一个Counter对应枚举类里面的一个枚举值。下面这段代码是Counter和CounterGroup的一个例子，在这个例子中，FileSystemCounter是一个枚举类，对应于FileSystemCounter这个CounterGroup，这个CounterGroup里面包含多个Counter，BYTES_READ是其中一个Counter。

public enum FileSystemCounter {
 BYTES_READ,
 BYTES_WRITTEN,
 READ_OPS,
 LARGE_READ_OPS,
 WRITE_OPS,
 BYTES_READ_EC,
}

Usage

Counter主要用来记录事件发生的次数。Hadoop mapreduce counter会追踪如下事件:

job里面包含的map/reduce任务数量
job read/write多少个字节
map/reduce launched/failed/killed count
cpu/memory usage of job

Counter types

在这里插入图片描述

总体上来讲，Counter分为built-in counter和customized counter两大类，built-in counter是mapreduce framework定义的counter，customized counter是用户在mapreduce application中自己定义的counter，下面我们以此介绍这些counter。

Built-in Counter

Built-in counter也就是hadoop mapreduce framework为每一个job定义的counter，用来上报和job相关的统计信息。举个例子，可以使用built-in counter来记录读/写了多个少字节，读/写了多少记录，通过这些统计信息我们可以确认，预期数量的输入被处理了，预期数量的输出被生产了。

TaskCounter

TaskCounter可以用来追踪每个task进度，是由每个TaskAttempt维护，会周期性的把counter发送给ApplicationMaster，然后在全局范围类做聚合。如下所示，我们会对比较重要的Counter的含义通过注释进行说明。

public enum TaskCounter {
  MAP_INPUT_RECORDS,
  MAP_OUTPUT_RECORDS,
  MAP_SKIPPED_RECORDS,
  MAP_OUTPUT_BYTES,
  MAP_OUTPUT_MATERIALIZED_BYTES,
  SPLIT_RAW_BYTES,
  COMBINE_INPUT_RECORDS,
  COMBINE_OUTPUT_RECORDS,
  REDUCE_INPUT_GROUPS,
  REDUCE_SHUFFLE_BYTES,
  REDUCE_INPUT_RECORDS,
  REDUCE_OUTPUT_RECORDS,
  REDUCE_SKIPPED_GROUPS,
  REDUCE_SKIPPED_RECORDS,
  SPILLED_RECORDS,
  SHUFFLED_MAPS, 
  failed_SHUFFLE,
  MERGED_MAP_OUTPUTS,
  GC_TIME_MILLIS,
  CPU_MILLISECONDS,
  PHYSICAL_MEMORY_BYTES,
  virtual_MEMORY_BYTES,
  COMMITTED_HEAP_BYTES,
  MAP_PHYSICAL_MEMORY_BYTES_MAX,
  MAP_VIRTUAL_MEMORY_BYTES_MAX,
  REDUCE_PHYSICAL_MEMORY_BYTES_MAX,
  REDUCE_VIRTUAL_MEMORY_BYTES_MAX;
}

Counter	Description
MAP_INPUT_RECORDS	记录每个map task处理得input record的个数，在RecordReader从InputSplit读取record并且传递给mapper的map方法的时候会增加
MAP_OUTPUT_RECORDS	记录每个map task产生的output record的个数，在调用Mapper.context类的write方法时会增加
REDUCE_INPUT_GROUPS	记录每个reduce task处理的key group的个数，在调用reduce方法的时候会增加
REDUCE_INPUT_RECORDS	记录每个reduce task处理的input record的个数，在通过Iterator对value进行迭代的过程中，没读取一个value，计数器数量会增加。如果reduce task处理了所有的input records，那么这个数量应该和MAP_OUTPUT_RECORDS相同
REDUCE_OUTPUT_RECORDS	记录每个reduce task产生的output record的个数，在调用Reducer.Context类的write方法时会增加
SPILLED_RECORDS	记录每个map/reduce task溢出记录的个数
CPU_MILLISECONDS	记录每个map/reduce task累计的cpu使用时间，来自/proc/cpuinfo
PHYSICAL_MEMORY_BYTES	记录每个map/reduce task累计使用的物理内存，来自/proc/meminfo
VIRTUAL_MEMORY_BYTES	记录每个map/reduce task累计使用的虚拟内存，来自/proc/meminfo

JobCounter

JobCounter可以追踪job的进度，JobCounter由ApplicationMaster来维护，所以这种类型的Counter不需要跨网络传输，然而其它所有类型的Counter，包括用户自定义的Counter都需要跨网络传输。

public enum JobCounter {
  NUM_FAILED_MAPS, 
  NUM_FAILED_REDUCES,
  NUM_KILLED_MAPS,
  NUM_KILLED_REDUCES,
  TOTAL_LAUNCHED_MAPS,
  TOTAL_LAUNCHED_REDUCES,
  OTHER_local_MAPS,
  DATA_LOCAL_MAPS,
  RACK_LOCAL_MAPS,
  @Deprecated
  SLOTS_MILLIS_MAPS,
  @Deprecated
  SLOTS_MILLIS_REDUCES,
  @Deprecated
  FALLOW_SLOTS_MILLIS_MAPS,
  @Deprecated
  FALLOW_SLOTS_MILLIS_REDUCES,
  TOTAL_LAUNCHED_UBERTASKS,
  NUM_UBER_SUBMAPS,
  NUM_UBER_SUBREDUCES,
  NUM_FAILED_UBERTASKS,
  TASKS_REQ_PREEMPT,
  checkpointS,
  CHECKPOINT_BYTES,
  CHECKPOINT_TIME,
  MILLIS_MAPS,
  MILLIS_REDUCES,
  VCORES_MILLIS_MAPS,
  VCORES_MILLIS_REDUCES,
  MB_MILLIS_MAPS,
  MB_MILLIS_REDUCES
}

Counter	Description
TOTAL_LAUNCHED_MAPS	记录每个job总共启动的map task的数量
TOTAL_LAUNCHED_REDUCES	记录每个job总共启动的reduce task的数量
NUM_UBER_SUBMAPS	记录每个uber job总共包括的map task的数量
NUM_UBER_SUBREDUCES	记录每个uber job总共包括的reduce task的数量
NUM_FAILED_MAPS	记录每个job总共失败的map task的数量
NUM_FAILED_REDUCES	记录每个job总共失败的reduce task的数量
NUM_KILLED_MAPS	记录每个job总共杀死的map task的数量
NUM_KILLED_REDUCES	记录每个job总共杀死的reduce task的数量
DATA_LOCAL_MAPS	记录每个job里面map task和input data在同一个节点的map task的数量
RACK_LOCAL_MAPS	记录每个job里面map task和input data在同一个rack但不在同一个node的map task的数量
OTHER_LOCAL_MAPS	记录每个job里面map task和input data不在同一个rack的map task的数量。不同rack之间的网络带宽是很稀缺的资源，hadoop尽可能的把map task放在input data的附近node上执行，所以这个值应该很小

FileSystemCounter

File system countres track 2 main details , number of bytes read by the file system and number of bytes written.

public enum FileSystemCounter {
  BYTES_READ,
  BYTES_WRITTEN,
  READ_OPS,
  LARGE_READ_OPS,
  WRITE_OPS,
  BYTES_READ_EC,
}

Counter	Description
BYTES_READ	每个map/reduce task在不同的文件系统上读的字节数
BYTES_WRITTEN	每个map/reduce task在不同的文件系统上写的字节数

fileinputformatCounter

public enum FileInputFormatCounter {
  BYTES_READ
}

Counter	Description
BYTES_READ	记录每个map task读取得字节数，在RecordReader从InputSplit读取record并且传递给mapper的map方法的时候会增加

FileOutputFormatCounter

public enum FileOutputFormatCounter {
  BYTES_WRITTEN
}

Counter	Description
BYTES_WRITTEN	记录每个reduce task(mapreduce job)或者map task(map only job)产生的字节数

Customized Counter

除了built-in的Counter之外，在mapreduce程序里面，可以定义自定义Counter，收集一些我们所关心的metrics。

Static Counter

一个mapreduce job可以定义任意数量的枚举类，每个枚举类可以定义任务数量的的枚举常量，每个枚举类对应一个CounterGroup，每个枚举常量对应一个Counter。

定义一个Counter一般分为两个步骤，第一步，定义枚举类和枚举常量；第二部，在Mapper/Reducer里面使用Counter

interface TaskAttemptContext {
  public Counter getCounter(Enum<?> counterName);
}

Dynamic Counter

通过定义枚举类和枚举常量的方式定义Counter，必须在程序运行之前把枚举类和枚举常量定义好，如果希望不事先定义枚举类和枚举常量，在程序运行的时候定义Counter，mapreduce framework也是支持的。只需要在获取Counter的时候，显示的指定CounterGroup名字和Counter名字，具体来说，通过下面这个方法事先。

interface TaskAttemptContext {
  public Counter getCounter(String groupName, String counterName);
}

Reference

https://data-flair.training/blogs/hadoop-counters/
https://acadgild.com/blog/counters-in-mapreduce

[MapReduce] Counter