MapReduce高级编程

今天药忘吃喽~ 2022-04-25 02:00 211阅读 0赞

MapReduce 计数器、最值：

## 计数器 ##

数据集在进行MapReduce运算过程中，许多时候，用户希望了解待分析的数据的运行的运行情况。Hadoop内置的计数器功能收集作业的主要统计信息，可以帮助用户理解程序的运行情况，辅助用户诊断故障。

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 </pre></td> 
 <td><pre>SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory] 18/12/28 10:37:46 INFO client.RMProxy: Connecting to ResourceManager at datanode3/192.168.1.103:8032 18/12/28 10:37:48 WARN mapreduce.JobResourceUploader: Hadoop command-line option parsing not performed. Implement the Tool interface and execute your application with ToolRunner to remedy this. 18/12/28 10:37:50 INFO input.FileInputFormat: Total input paths to process : 2 18/12/28 10:37:50 INFO mapreduce.JobSubmitter: number of splits:2 18/12/28 10:37:51 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1545964109134_0001 18/12/28 10:37:53 INFO impl.YarnClientImpl: Submitted application application_1545964109134_0001 18/12/28 10:37:54 INFO mapreduce.Job: The url to track the job: http://datanode3:8088/proxy/application_1545964109134_0001/ 18/12/28 10:37:54 INFO mapreduce.Job: Running job: job_1545964109134_0001 18/12/28 10:38:50 INFO mapreduce.Job: Job job_1545964109134_0001 running in uber mode : false 18/12/28 10:38:50 INFO mapreduce.Job: map 0% reduce 0% 18/12/28 10:39:28 INFO mapreduce.Job: map 100% reduce 0% 18/12/28 10:39:48 INFO mapreduce.Job: map 100% reduce 100% 18/12/28 10:39:50 INFO mapreduce.Job: Job job_1545964109134_0001 completed successfully 18/12/28 10:39:51 INFO mapreduce.Job: Counters: 49 File System Counters FILE: Number of bytes read=78 FILE: Number of bytes written=353015 FILE: Number of read operations=0 FILE: Number of large read operations=0 FILE: Number of write operations=0 HDFS: Number of bytes read=258 HDFS: Number of bytes written=31 HDFS: Number of read operations=9 HDFS: Number of large read operations=0 HDFS: Number of write operations=2 Job Counters Launched map tasks=2 Launched reduce tasks=1 Data-local map tasks=2 Total time spent by all maps in occupied slots (ms)=67297 Total time spent by all reduces in occupied slots (ms)=16699 Total time spent by all map tasks (ms)=67297 Total time spent by all reduce tasks (ms)=16699 Total vcore-milliseconds taken by all map tasks=67297 Total vcore-milliseconds taken by all reduce tasks=16699 Total megabyte-milliseconds taken by all map tasks=68912128 Total megabyte-milliseconds taken by all reduce tasks=17099776 Map-Reduce Framework Map input records=8 Map output records=8 Map output bytes=78 Map output materialized bytes=84 Input split bytes=212 Combine input records=8 Combine output records=6 Reduce input groups=4 Reduce shuffle bytes=84 Reduce input records=6 Reduce output records=4 Spilled Records=12 Shuffled Maps =2 Failed Shuffles=0 Merged Map outputs=2 GC time elapsed (ms)=3303 CPU time spent (ms)=8060 Physical memory (bytes) snapshot=470183936 Virtual memory (bytes) snapshot=6182424576 Total committed heap usage (bytes)=261361664 Shuffle Errors BAD_ID=0 CONNECTION=0 IO_ERROR=0 WRONG_LENGTH=0 WRONG_MAP=0 WRONG_REDUCE=0 File Input Format Counters Bytes Read=46 File Output Format Counters Bytes Written=31 </pre></td> 
 </tr> 
 </tbody> 
</table>

这些记录了该程序运行过程的的一些信息的计数，如Map input records=8，表示Map有8条记录。可以看出来这些内置计数器可以被分为若干个组，即对于大多数的计数器来说，Hadoop使用的组件分为若干类。

### 计数器列表 ###

<table> 
 <thead> 
 <tr> 
 <th>组别</th> 
 <th>名称/类别</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td>MapReduce任务计数器（Map-Reduce Framework）</td> 
 <td>org.apache.hadoop.mapreduce.TaskCounter</td> 
 </tr> 
 <tr> 
 <td>文件系统计数器（File System Counters）</td> 
 <td>org.apache.hadoop.mapreduce.FiIeSystemCounter</td> 
 </tr> 
 <tr> 
 <td>输入文件任务计数器（File Input Format Counters）</td> 
 <td>org.apache.hadoop.mapreduce.lib.input.FilelnputFormatCounter</td> 
 </tr> 
 <tr> 
 <td>输出文件计数器（File Output Format Counters）</td> 
 <td>org.apache.hadoop.mapreduce.lib.output.FileOutputFormatCounter</td> 
 </tr> 
 <tr> 
 <td>作业计数器（Job Counters）</td> 
 <td>org.apache.hadoop.mapreduce.JobCounter</td> 
 </tr> 
 </tbody> 
</table>

大部分的Hadoop都有相应的计数器，可以对其进行追踪，方便处理运行中出现的问题，这些信息从应用角度又分为任务计数器和作业计数器：

#### 任务计数器 ####

##### 内置MapReduce任务计数器 #####

<table> 
 <thead> 
 <tr> 
 <th>计数器名称</th> 
 <th>说明</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td>map输人的记录数(MAP_INPUT_RECORDS）</td> 
 <td>作业中所有map已处理的输人记录数。每次RecordReader读到一条记录并将其传给map的map()函数时，该计数器的值递增</td> 
 </tr> 
 <tr> 
 <td>分片（split）的原始字节数(SPLIT_RAW_BYTES)</td> 
 <td>由map读取的输人分片对象的字节数。这些对象描述分片元数据（文件的位移和长度），而不是分片的数据自身，因此总规模是小的</td> 
 </tr> 
 <tr> 
 <td>map输出的记录数(MAP_OUTPUT_RECORDS)</td> 
 <td>作业中所有map产生的map输出记录数。每次某一个map 的OutputCollector调用collect()方法时，该计数器的值增加</td> 
 </tr> 
 <tr> 
 <td>map输出的字节数(MAP_OUTPUT_BYTES)</td> 
 <td>作业中所有map产生的耒经压缩的输出数据的字节数·每次某一个map的OutputCollector调用collect()方法时，该计数器的值增加</td> 
 </tr> 
 <tr> 
 <td>map输出的物化字节数（MAP_OUTPUT_MATERIALIZED_BYTES)</td> 
 <td>map输出后确实写到磁盘上的字节数；若map输出压缩功能被启用，则会在计数器值上反映出来</td> 
 </tr> 
 <tr> 
 <td>combine输人的记录数(COMBINE_INPUT_RECORDS)</td> 
 <td>作业中所有combiner(如果有）已处理的输人记录数。combiner的迭代器每次读一个值，该计数器的值增加。注意：本计数器代表combiner已经处理的值的个数，并非不同的键组数（后者并无实所意文，因为对于combiner而言，并不要求每个键对应一个组。</td> 
 </tr> 
 <tr> 
 <td>combine输出的记录数(COMBINE_OUTPUT_RECORDS)</td> 
 <td>作业中所有combiner（如果有）已产生的输出记录数。每当一个combiner的OutputCollector调用collect()方法时，该计数器的值增加</td> 
 </tr> 
 <tr> 
 <td>reduce输人的组（REDUCE_INPUT_GROUPS）</td> 
 <td>作业中所有reducer已经处理的不同的码分组的个数。每当某一个reducer的reduce()被调用时，该计数器的值增加。</td> 
 </tr> 
 <tr> 
 <td>reduce输人的记录数（REDUCE_INPUT_RECORDS)</td> 
 <td>作业中所有reducer已经处理的输人记录的个数。每当某个reducer的迭代器读一个值时，该计数器的值增加。如果所有reducer已经处理数完所有输人，則该计数器的值与计数器”map输出的记录”的值相同。</td> 
 </tr> 
 <tr> 
 <td>reduce输出的记录数（REDUCE_OUTPUT_RECORDS）</td> 
 <td>作业中所有map已经产生的reduce输出记录数。每当某个reducer的OutputCollector调用collect()方法时，该计数器的值增加。</td> 
 </tr> 
 <tr> 
 <td>reduce经过shuffle的字节数(REDUCE_SHUFFLE_BYTES)</td> 
 <td>由shuffle复制到reducer的map输出的字节数。</td> 
 </tr> 
 <tr> 
 <td>溢出的记录数(SPILLED_RECORDS)</td> 
 <td>作业中所有map和reduce任务溢出到磁盘的记录数</td> 
 </tr> 
 <tr> 
 <td>CPU毫秒(CPU_MILLISECONDS)</td> 
 <td>一个任务的总CPU时间，以毫秒为单位，可由/proc/cpuinfo获取</td> 
 </tr> 
 <tr> 
 <td>物理内存字节数（PHYSICAL_MEMORY_BYTES）</td> 
 <td>一个任务所用的物理内存，以字节数为单位，可由/proc/meminfo获取</td> 
 </tr> 
 <tr> 
 <td>虚拟内存字节数(VIRTUAL_MEMORY_BYTES）</td> 
 <td>一个任务所用虚拟内存的字节数，由/proc/meminfo获取</td> 
 </tr> 
 <tr> 
 <td>有效的堆字节数(COMMITTED_HEAP_BYTES)</td> 
 <td>在JVM中的总有效内存最（以字节为单位），可由Runtime. getRuntime().totalMemory()获取</td> 
 </tr> 
 <tr> 
 <td>GC运行时间毫秒数(GC_TIME_MILLIS)</td> 
 <td>在任务执行过程中，垃圾收集器(garbage collection）花费的时间（以毫秒为单位），可由GarbageCollector MXBean. getCollectionTime()获取</td> 
 </tr> 
 <tr> 
 <td>由shuffle传输的map输出数(SHUFFLED_MAPS)</td> 
 <td>由shume传输到reducer的map输出文件数。</td> 
 </tr> 
 <tr> 
 <td>失敗的shuffle数(FAILED_SHUFFLE)</td> 
 <td>shuffle过程中，发生map输出拷贝错误的次数</td> 
 </tr> 
 <tr> 
 <td>被合并的map输出数（MERGED_MAP_OUTPUTS）</td> 
 <td>shuffle过程中，在reduce端合并的map输出文件数</td> 
 </tr> 
 </tbody> 
</table>

##### 内置文件系统任务计数器 #####

<table> 
 <thead> 
 <tr> 
 <th>计数器名称</th> 
 <th>说明</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td>文件系统的读字节数（BYTES_READ）</td> 
 <td>由map任务和reduce任务在各个文件系统中读取的字节数，各个文件系统分别对应一个计数器，文件系统可以是local、 HDFS、S3等</td> 
 </tr> 
 <tr> 
 <td>文件系统的写字节数(BYTES_WRITTEN）</td> 
 <td>由map任务和reduce任务在各个文件系统中写的字节数</td> 
 </tr> 
 <tr> 
 <td>文件系统读操作的数量(READ_OPS)</td> 
 <td>由map任务和reduce任务在各个文件系统中进行的读操作的数量（例如，open操作，filestatus操作）</td> 
 </tr> 
 <tr> 
 <td>文件系统大规模读操作的数量(LARGE_READ_OPS)</td> 
 <td>由map和reduce任务在各个文件系统中进行的大规模读操作（例如，对于一个大容量目录进行list操作）的数量</td> 
 </tr> 
 <tr> 
 <td>文件系统写操作的数量(WRITE_OPS)</td> 
 <td>由map任务和reduce任务在各个文件系统中进行的写操作的数量（例如，create操作，append操作）</td> 
 </tr> 
 </tbody> 
</table>

##### 内置的输入文件任务计数器 #####

<table> 
 <thead> 
 <tr> 
 <th>计数器名称</th> 
 <th>说明</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td>读取的字节数(BYTES_READ)</td> 
 <td>由map任务通过FilelnputFormat读取的字节数</td> 
 </tr> 
 </tbody> 
</table>

##### 内置输出文件任务计数器 #####

<table> 
 <thead> 
 <tr> 
 <th>计数器名称</th> 
 <th>说明</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td>写的字节数(BYTES_WRITTEN)</td> 
 <td>由map任务（针对仅含map的作业）或者reduce任务通过FileOutputFormat写的字节数</td> 
 </tr> 
 </tbody> 
</table>

#### 作业计数器 ####

##### 内置的作业计数器 #####

<table> 
 <thead> 
 <tr> 
 <th>计数器名称</th> 
 <th>说明</th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td>启用的map任务数（TOTAL_LAUNCHED_MAPS）</td> 
 <td>启动的map任务数，包括以“推测执行”方式启动的任务。</td> 
 </tr> 
 <tr> 
 <td>启用的reduce任务数(TOTAL_LAUNCHED_REDUCES)</td> 
 <td>启动的reduce任务数，包括以“推测执行”方式启动的任务。</td> 
 </tr> 
 <tr> 
 <td>启用的uber任务数(TOTAL_LAIÆHED_UBERTASKS)</td> 
 <td>启用的uber任务数。</td> 
 </tr> 
 <tr> 
 <td>uber任务中的map数(NUM_UBER_SUBMAPS)</td> 
 <td>在uber任务中的map数。</td> 
 </tr> 
 <tr> 
 <td>Uber任务中的reduce数(NUM_UBER_SUBREDUCES)</td> 
 <td>在任务中的reduce数。</td> 
 </tr> 
 <tr> 
 <td>失败的map任务数（NUM_FAILED_MAPS）</td> 
 <td>失败的map任务数。</td> 
 </tr> 
 <tr> 
 <td>失败的reduce任务数(NUM_FAILED_REDUCES)</td> 
 <td>失败的reduce任务数</td> 
 </tr> 
 <tr> 
 <td>失败的uber任务数(NIN_FAILED_UBERTASKS)</td> 
 <td>失败的uber任务数。</td> 
 </tr> 
 <tr> 
 <td>被中止的map任务数（NUM_KILLED_MAPS）</td> 
 <td>被中止的map任务数。</td> 
 </tr> 
 <tr> 
 <td>被中止的reduce任务数(NW_KILLED_REDUCES)</td> 
 <td>被中止的reduce任务数。</td> 
 </tr> 
 <tr> 
 <td>数据本地化的map任务数（DATA_LOCAL_MAPS）</td> 
 <td>与输人数据在同一节点上的map任务数。</td> 
 </tr> 
 <tr> 
 <td>机架本地化的map任务数（RACK_LOCAL_MAPS)</td> 
 <td>与输人数据在同一机架范围内但不在同一节点上的map任务数。</td> 
 </tr> 
 <tr> 
 <td>其他本地化的map任务数（OTHER_LOCAL_MAPS）</td> 
 <td>与输人数据不在同一机架范围内的map任务数。由于机架之间的带宽资源相对较少，Hadoop会尽量让map任务靠近输人数据执行，因此该计数器值一般比较小。</td> 
 </tr> 
 <tr> 
 <td>map任务的总运行时间(MILLIS_MAPS)</td> 
 <td>map任务的总运行时间，单位毫秒。包括以推测执行方式启动的任务。可参见相关的度量内核和内存使用的计数器(VCORES_MILLIS_MAPS和MB_MILLIS_MAPS）</td> 
 </tr> 
 <tr> 
 <td>reduce任务的总运行时间(MILLIS_REDUCES)</td> 
 <td>reduce任务的总运行时间，单位毫秒。包括以推滌执行方式启动的任务。可参见相关的度量内核和内存使用的计数器(VQES_MILLIS_REARES和t*B_MILLIS_REUKES)</td> 
 </tr> 
 </tbody> 
</table>

#### 自定义计数器 ####

虽然Hadoop内置的计数器比较全面，给作业运行过程的监控带了方便，但是对于那一些业务中的特定要求(统计过程中对某种情况发生进行计数统计)MapReduce还是提供了用户编写自定义计数器的方法。

##### 过程 #####

1.  定义一个Java的枚举类型(enum)，用于记录计数器分组，其枚举类型的名称即为分组的名称，枚举类型的字段就是计数器名称。
2.  通过Context类的实例调用getCounter方法进行increment(long incr)方法，进行计数的添加。

##### 案例 #####

###### ReportTest ######

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 </pre></td> 
 <td><pre>public enum ReportTest { //定义枚举 ErroWord, GoodWord, ReduceReport	//写入要记录的计数器名称 } </pre></td> 
 </tr> 
 </tbody> 
</table>

###### Mapper类 ######

###### Reducer类 ######

###### ToolRunnerJS ######

#### 查看 ####

![FWk6ln.png][]

通过Web界面也可以查看但是需要设置设置

启动服务

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 </pre></td> 
 <td><pre>mr-jobhistory-daemon.sh start historyserver </pre></td> 
 </tr> 
 </tbody> 
</table>

web界面查看

![FWk5Y4.png][]

## 最值 ##

最大值、最小值、平均值、均方差、众数、中位数等都是统计学中经典的数值统计，也是常用的统计属性字段，如果想知道最大的10个数，最小的10个数，这涉及到Top N/Bottom N 问题。

### 单一最值 ###

常用的统计属性的字段在MapReduce的求解过程中，由一个大任务分解成若干个Mapper任务，最后会进行Reducer合并，比传统计算求解略显复杂，在MaoReduce框架中，会以Key进行分区、分组、排序的操作，在进行这些数值的操作时哦，只要设定合理的key，整个问题也就简单化了。使用Combiner可以减少Shuffle到Reduce端中间的K V的数目，减轻网络和IO的目的。

#### 求解最大值最小值 ####

数据

#### MinMaxWritable ####

#### MinMaxMapper ####

#### MinMaxReducer ####

#### MinMaxJob ####

#### 计算过程 ####

![FfaTAK.png][]

在一个MapReduce计算的过程中，Mapper任务相对于Reduce任务是大量的，因此少量的Reducer处理大量数据的并不明智，所以通过在Shuffle阶段引入Combiner，并把Reducer作为它的计算类，大大减少了Reducer端数据的输入，整个计算过程变得合理可靠。

[FWk6ln.png]: /images/20220210/f3757dcc8dbe4ef29baae01563198e2a.png
[FWk5Y4.png]: /images/20220210/d01160c469dd4820bb2561bfa58941b5.png
[FfaTAK.png]: /images/20220210/6c02e5bfc477475cb45f8ddecc9aaa2b.png