Flume-NG内置计数器(监控)源码级分析

  Flume的内置监控怎么整?这个问题有很多人问。目前了解到的信息是可以使用Cloudera Manager、Ganglia有图形的监控工具,以及从浏览器获取json串,或者自定义向其他监控系统汇报信息。那监控的信息是什么呢?就是各个组件的统计信息,比如成功接收的Event数量、成功发送的Event数量,处理的Transaction的数量等等。而且不同的组件有不同的Countor来做统计,目前直到1.5版本仍然只对三大组件:source、sink、channel进行统计分别是SourceCounter、SinkCounter、ChannelCounter,这三个计数器的统计项是固定的,就是你不能自己设置自己的统计项;另外还有ChannelProcessorCounter和SinkProcessorCounter,这两项目前没有设置统计项,所以是目前还是“摆设”。另外有些同学可能也发现了,有些内置的组件使用CounterGroup这个来统计信息,这个可以自己随意设置统计项,但是遗憾的是目前(1.5版本)这个可以自定义的计数器的信息还无法用在监控上,因为这只是一个单独的类,并没有继承MonitoredCounterGroup这个抽象类。有些内置组件使用的是CounterGroup,所以监控时会没有数据,不同的版本使用此CounterGroup的组件可能不同。下面我们重点介绍:SourceCounter、SinkCounter、ChannelCounter。

  Flume-NG的所有统计信息、监控及相关的类都在org.apache.flume.instrumentation.http、org.apache.flume.instrumentation、org.apache.flume.instrumentation.util三个包下。

  上面提到了MonitoredCounterGroup,这个类是用来跟踪内部的统计指标的,注册组件的MBean并跟踪和更新统计值。需要监控的组件都要继承这个类,这个类可以跟踪flume内部的所有组件,但是目前只实现了3个。其中比较重要的方法有以下几个:

  (1)、构造方法MonitoredCounterGroup(Type type, String name, String... attrs),这个方法主要是设置组件的类型、名称;然后将所有的attrs(这是设定的各个统计项)加入Map<String, AtomicLong> counterMap,值设定为0;然后初始化计数器的开始时间和结束时间,都设为0.

  (2)、start()方法,会先注册计数器,然后对所有统计项的统计值设为0;将开始时间设置为当前时间

  (3)、register()方法,如果这个计数器还未注册,将这个计数器的MBean进行注册,就可以进行跟踪了

  (4)、stop()方法,会设置结束时间为当前时间;输出各个统计项的信息。我们 Ctrl+C 结束进程时,最后显示的统计信息就是来自这里。

  其它方法都是获取counterMap的中信息或者更新值等,比较简单。

  接下来我们看看,三个组件中各种统计项及其含义吧:

  一、SourceCounter,继承了MonitoredCounterGroup。主要统计项如下:

  (1)"src.events.received",表示source接受的event个数;

  (2)"src.events.accepted",表示source处理成功的event个数,和上面的区别就是上面虽然接受了可能没处理成功;

  (3)"src.append.received",表示调用append次数,在avrosource和thriftsource中调用;

  (4)"src.append.accepted",表示append处理成功次数;

  (5)"src.append-batch.received",表示appendBatch被调用的次数,在avrosource和thriftsource中调用;

  (6)"src.append-batch.accepted",表示appendBatch处理成功次数;

  (7)"src.open-connection.count",用在avrosource中表示打开连接的数量;

  一般source调用都集中在前俩。

  二、SinkCounter,继承了MonitoredCounterGroup

  (1)"sink.connection.creation.count",这个调用的地方颇多,都表示“链接”创建的数量,比如与HBase建立链接,与avrosource建立链接以及文件的打开等;

  (2)"sink.connection.closed.count",对应于上面的stop操作、destroyConnection、close文件操作等。

  (3)"sink.connection.failed.count",表示上面所表示“链接”时异常、失败的次数;

  (4)"sink.batch.empty",表示这个批次处理的event数量为0的情况;

  (5)"sink.batch.underflow",表示这个批次处理的event的数量介于0和配置的batchSize之间;

  (6)"sink.batch.complete",表示这个批次处理的event数量等于设定的batchSize;

  (7)"sink.event.drain.attempt",准备处理的event的个数;

  (8)"sink.event.drain.sucess",这个表示处理成功的event数量,与上面不同的是上面的是还未处理的。

  三、ChannelCounter,继承了MonitoredCounterGroup

  (1)"channel.current.size",这个表示这个channel的当前容量;

  (2)"channel.event.put.attempt",一般指的是在channel的事务当中,source的put操作中记录尝试发送event的个数;

  (3)"channel.event.take.attempt",一般指的是在channel的事务中,sink的take操作记录尝试拿event的个数;

  (4)"channel.event.put.success",一般指的是在channel的事务中,put成功的event的数量;

  (5)"channel.event.take.success",一般指的是channel事务中,take成功的event的数量;

  (6)"channel.capacity",指的是channel的容量,在channel的start方法中设置。

  上面这些统计项都是固定的,我们可以根据需要增加相应项的值,可以在监控中查看组件的变化情况,从而掌握flume进程的运行情况。比如可以查看channel的容量从而了解到source和sink的相对处理速度,还有可以看source或者sink每个批次处理成功与失败的次数,了解组件的运行状况等等。

  当然有些同学可能在自定义自己的组件时,想统计一些自己的统计项,这些统计项在上面三大组件中是没有,怎么办?自己定制啊,上面说了必须要继承MonitoredCounterGroup这个抽象类,设定自己的统计项,然后将统计项设置成数组调用MonitoredCounterGroup的构造函数;然后在自定义的计数器中增加更新数值的方法。最后在自定义的组件中构造自定义的计数器,并启用它的start方法,剩下的就是在该更新统计项数值的地方更新就可以了。

  还有一个重要的内容就是监控的实现!没错,内置的有两种HTTP方式(就是json串)和Ganglia,后者需要安装Ganglia,前者非常简单,只需要在Flume的启动命令中加上:-Dflume.monitoring.type=http -Dflume.monitoring.port=XXXX  ,最后的XXXX是你需要设置的端口!然后你就可以在浏览器上通过访问这个Flume所在节点的IP:XXXX/metrics,不断刷新就可以看到最新的组件统计信息。关于Ganglia的请读者自行组件Ganglia集群并餐卡用户指南来操作。

  如果我想自己实现一个server向其他系统汇报信息,咋整?目前有至少两个方法:

  一、就是上面的HTTP啊,你可以不断去获取json串,自己解析出来各个统计指标,然后剩下的就是你想怎么整就怎么整吧。

  二、就是自己实现一个类似HTTP的server,必须实现org.apache.flume.instrumentation.MonitorService接口,这个接口只有俩方法:start和stop。这个接口继承自Configurable接口所以拥有可以读取配置文件的configure(configure(Context context))方法,来获取一些配置信息。

  以HTTP为例(对应的类是org.apache.flume.instrumentation.http.HTTPMetricsServer),它的start方法启动了一个jetty作为web server,提供WEB服务。并实现了AbstractHandler的一个处理数据的类HTTPMetricsHandler,这个类的handle(String target, HttpServletRequest request, HttpServletResponse response,int dispatch)方法来设置一些WEB页面的格式以及通过JMXPollUtil.getAllMBeans()获取所有组件注册的MBean构成的Map<String, Map<String, String>> metricsMap,遍历这个metricsMap将这个metricsMap转换成json输出到web页面。stop方法就是一些清理工作,这里是关闭jetty server。很简单吧,所以我们完全可以实现一个server,在start方法中启动一个线程每隔一秒或者自己定遍历这个metricsMap,写入mysql、HBase或者别的地方,你随便。。。

  你可以在定义的组件中调用自己的计数器,然后将计数器、监控类、自定义组件(source、sink、channel)打包放到lib下,在启动命令后加-Dflume.monitoring.type=AAAAA -Dflume.monitoring.node=BBBB,就可以了。注意,Dflume.monitoring.type这个好似必须要设置的,就是你自己的监控类(这里是AAAAA),后面的可有可无都是一些参数,你可以自定义参数名,比如可以设置数据库服务器IP、端口等。

    至此,这里介绍完了。这些都是从源码中看出来的,还未曾实现,供大伙借鉴。

Flume-NG内置计数器(监控)源码级分析

时间: 2025-01-07 06:42:05

Flume-NG内置计数器(监控)源码级分析的相关文章

Matlab.NET混合编程技巧之——直接调用Matlab内置函数(附源码)

原文:[原创]Matlab.NET混合编程技巧之--直接调用Matlab内置函数(附源码) 在我的上一篇文章[原创]Matlab.NET混编技巧之--找出Matlab内置函数中,已经大概的介绍了matlab内置函数在混合编程中的优点,并通过程序找出了matlab中的大部分内置函数,当然更多人关心是如何像我所说得那样,不用直接编译,就直接在C#中调用这些内置函数.本文就带你揭开这些谜团. 声明,这篇文章是需要一点点混合编程基础的,基本概念和过程要懂一点,如果能简单成功混编一个简单的计算或者绘图例子

Flume-NG(1.5版本)中SpillableMemoryChannel源码级分析

SpillableMemoryChannel是1.5版本新增的一个channel.这个channel优先将evnet放在内存中,一旦内存达到设定的容量就使用file channel写入磁盘.然后读的时候会按照顺序读取:会通过一个DrainOrderQueue来保证不管是内存中的还是溢出(本文的“溢出”指的是内存channel已满,需要使用file channel存储数据)文件中的顺序.这个Channel是memory channel和file channel的一个折中,虽然在内存中的数据仍然可能

MapReduce job在JobTracker初始化源码级分析

mapreduce job提交流程源码级分析(三)中已经说明用户最终调用JobTracker.submitJob方法来向JobTracker提交作业.而这个方法的核心提交方法是JobTracker.addJob(JobID jobId, JobInProgress job)方法,这个addJob方法会把Job提交到调度器(默认是JobQueueTaskScheduler)的监听器JobQueueJobInProgressListener和EagerTaskInitializationListen

监听器初始化Job、JobTracker相应TaskTracker心跳、调度器分配task源码级分析

JobTracker和TaskTracker分别启动之后(JobTracker启动流程源码级分析,TaskTracker启动过程源码级分析),taskTracker会通过心跳与JobTracker通信,并获取分配它的任务.用户将作业提交到JobTracker之后,放入相应的数据结构中,静等被分配.mapreduce job提交流程源码级分析(三)这篇文章已经分析了用户提交作业的最后步骤,主要是构造作业对应的JobInProgress并加入jobs,告知所有的JobInProgressListen

TableInputFormat分片及分片数据读取源码级分析

我们在MapReduce中TextInputFormat分片和读取分片数据源码级分析 这篇中以TextInputFormat为例讲解了InputFormat的分片过程以及RecordReader读取分片数据的过程.接下来咱们分析TableInputFormat的分片信息和数据读取过程. TableInputFormat这是专门处理基于HBase的MapReduce的输入数据的格式类.我们可以看看继承结构:(1)public class TableInputFormat extends Table

Shell主要逻辑源码级分析(1)——SHELL运行流程

版权声明:本文由李航原创文章,转载请注明出处: 文章原文链接:https://www.qcloud.com/community/article/109 来源:腾云阁 https://www.qcloud.com/community 本文的目的:分享一下在学校的时候分析shell源码的一些收获,帮助大家了解shell的一个工作流程,从软件设计的角度,看看shell这样一个历史悠久的软件的一些设计优点和缺陷.本文重点不是讲SHELL语法,相信很多同事玩shell都很熟了. 本文的局限:限于本人技术水

源码级强力分析Hadoop的RPC机制

分析对象: hadoop版本:hadoop 0.20.203.0 必备技术点: 1. 动态代理(参考 :http://weixiaolu.iteye.com/blog/1477774 )2. Java NIO(参考 :http://weixiaolu.iteye.com/blog/1479656 )3. Java网络编程 目录: 一.RPC协议二.ipc.RPC源码分析三.ipc.Client源码分析四.ipc.Server源码分析 分析:  一.RPC协议 在分析协议之前,我觉得我们很有必要先

MapReduce中TextInputFormat分片和读取分片数据源码级分析

InputFormat主要用于描述输入数据的格式(我们只分析新API,即org.apache.hadoop.mapreduce.lib.input.InputFormat),提供以下两个功能: (1)数据切分:按照某个策略将输入数据切分成若干个split,以便确定MapTask个数以及对应的split: (2)为Mapper提供输入数据:读取给定的split的数据,解析成一个个的key/value对,供mapper使用. InputFormat有两个比较重要的方法:(1)List<InputSp

9.1hadoop 内置计数器、自定义枚举计数器、Streaming计数器

1.1  计数器 计数器的作用是用来统计数量的,用于记录特定事件的次数,分为内置计数器.自定义java枚举计数器.自定义Stream计数器三大类.用于质量分析,或应用级统计.分析计数器的值比分析一堆日志更高效. 计数器名称 计数器介绍 内置计数器 Hadoop自带的计数器,有特定的计数器名称.例如统计输入.输出的记录数量,输入输出的字节数. 自定义java枚举计数器 用户自定义的枚举型计数器,用于统计用户的特殊要求的计数器,例如统计记录中无效记录的数量. 自定义Streaming计数器 通过向标