对ETL系统中数据转换和存储操作的相关日志进行记录以及实时分析有助于我们更好的观察和监控ETL系统的相关指标(如单位时间某些操作的处理时间),发现系统中出现的缺陷和性能瓶颈。
由于需要对日志进行实时分析,所以Storm是我们想到的首个框架。Storm是一个分布式实时计算系统,它可以很好的处理流式数据。利用storm我们几乎可以直接实现一个日志分析系统,但是将日志分析系统进行模块化设计可以收到更好的效果。模块化的设计至少有两方面的优点:
Flume是一个高可用、高可靠、分布式的海量日志采集、聚合和传输系统。Flume支持在日志系统中定制日志发送方,用于收集数据;同时,Flume提供对数据进行简单处理,并写到各种数据接收方的能力。它拥有一个简单的、可扩展的流式数据流架构,如下图所示:
日志收集系统就是由一个或者多个agent(代理)组成,每个agent由source、channel、sink三部分组成,source是数据的来源,channel是数据进行传输的通道,sink用于将数据传输到指定的地方。我们可以把agent看做一段水管,source是水管的入口,sink是水管的出口,数据流就是水流。 Agent本质上是一个jvm进程,agent各个组件之间是通过event来进行触发和协调的。
解压缩,并在conf目录下面新建一个文件flume-conf.properties,内容如下:
启动代理。flume-ng agent –n a1 –f flume-conf.properties
Kafka是linkedin用于日志处理的分布式消息队列。Kafka的架构如下图所示:
Kafka的存储策略有一下几点:
Kafka集群的搭建需要依赖zookeeper来进行负载均衡,所以我们需要在安装kafka之前搭建zookeeper集群。
Storm是一个分布式的、高容错的实时计算系统。Storm对于实时计算的的意义相当于Hadoop对于批处理的意义。hadoop为我们提供了Map和Reduce原语,使我们对数据进行批处理变的非常的简单和优美。同样,Storm也对数据的实时计算提供了简单Spout和Bolt原语。
Strom集群里面有两种节点,控制节点和工作节点,控制节点上面运行一个nimbus(类似于hadoop中的JobTracker)后台程序,Nimbus负责在集群里面分布代码,分配工作给机器, 并且监控状态。每一个工作节点上面运行一个叫做Supervisor(类似Hadoop中的TaskTracker)的节点。Supervisor会监听分配给它那台机器的工作,根据需要启动/关闭工作进程。每一个工作进程执行一个Topology(类似hadoop中的Job)的一个子集;一个运行的Topology由运行在很多机器上的很多工作进程 Worker(类似Hadoop中的Child)组成。结构如下图所示:
Stream是storm里面的关键抽象。一个stream是一个没有边界的tuple序列。storm提供一些原语来分布式地、可靠地把一个stream传输进一个新的stream。比如: 你可以把一个tweets流传输到热门话题的流。
storm提供的最基本的处理stream的原语是spout和bolt。你可以实现Spout和Bolt对应的接口以处理你的应用的逻辑。
Spout是流的源头。比如一个spout可能从Kestrel队列里面读取消息并且把这些消息发射成一个流。通常Spout会从外部数据源(队列、数据库等)读取数据,然后封装成Tuple形式,之后发送到Stream中。Spout是一个主动的角色,在接口内部有个nextTuple函数,Storm框架会不停的调用该函数。
Bolt可以接收任意多个输入stream。Bolt处理输入的Stream,并产生新的输出Stream。Bolt可以执行过滤、函数操作、Join、操作数据库等任何操作。Bolt是一个被动的角色,其接口中有一个execute(Tuple input)方法,在接收到消息之后会调用此函数,用户可以在此方法中执行自己的处理逻辑。
spout和bolt所组成一个网络会被打包成topology, topology是storm里面最高一级的抽象(类似 Job), 你可以把topology提交给storm的集群来运行。Topology的结构如下图所示:
Storm集群的搭建也要依赖于zookeeper,本系统中storm与kafka共用同样一个zookeeper集群。
配置nimbus。 修改storm的conf/storm.yaml文件如下:
注意:在每个配置项前面必须留有空格,否则会无法识别。storm.messaging.* 部分是Netty的配置。如果没有该部分。那么Storm默认还是使用ZeroMQ。
配置supervisor 修改storm的conf/storm.yaml文件如下:
注意
Drools是一个基于Java的、开源的规则引擎,可以将复杂多变的规则从硬编码中解放出来,以规则脚本的形式存放在文件中,使得规则的变更不需要修正代码重启机器就可以立即在线上环境生效。 日志分析系统中,drools的作用是利用不同的规则对日志信息进行处理,以获得我们想要的数据。但是,Drools本身不是一个分布式框架,所以规则引擎对log的处理无法做到分布式。我们的策略是将drools整合到storm的bolt中去,这就就解决了drools无法分布式的问题。这是因为bolt可以作为task分发给多个worker来处理,这样drools中的规则也自然被多个worker处理了。
Redis是key-value存储系统,它支持较为丰富的数据结构,有String,list,set,hash以及zset。与memcached一样,为了保证效率,数据都是缓存在内存中。区别的是redis会周期性的把更新的数据写入磁盘或者把修改操作写入追加的记录文件,并且在此基础上实现了master-slave(主从)同步。 Redis是内存数据库,所以有非常快速的存取效率。日志分析系统数据量并不是特别大,但是对存取的速度要求较高,所以选择redis有很大的优势。
Flume如何收集ETL系统中的日志是我需要考虑的第一个问题。log4j2提供了专门的Appender-FlumeAppender用于将log信息发送到flume系统,并不需要我们来实现。我们在log4j2的配置文件中配置了ETL系统将log信息发送到的目的地,即avro服务器端。该服务器端我们在flume的配置文件中进行了配置。配置信息如下所示:
我们从ETL系统中获得了日志信息,将该信息不作任何处理传递到sink端,sink端发送数据到kafka。这个发送过程需要我们编写代码来实现,我们的实现代码为KafkaSink类。主要代码如下所示:
该类中,我们读取了一些配置信息,这些配置信息我们在flumeng的flume-conf.properties文件中进行了定义,定义内容如下:
将上面的KafkaSink类打包成flumeng-kafka.jar,并将该jar包以及kafka_2.9.2-0.8.1.jar、metrics-annotation-2.2.0.jar、metrics-core-2.2.0.jar、Scala-compiler.jar、scala-library.jar、zkclient-0.3.jar放到flume的lib目录下,启动flume,我们就可以将ETL系统中产生的日志信息发送到kafka中的fks1这个topic中去了。
Storm中的spout如何主动消费kafka中的消息需要我们编写代码来实现,httpsgithub.comwurstmeisterstorm-kafka-0.8-plus实现了一个kafka与storm整合的插件,下载该插件,将插件中的jar包以及metrics-core-2.2.0.jar、scala-compiler2.9.2.jar放到storm的lib目录下。利用插件中的StormSpout类,我们就可以消费kafka中的消息了。主要代码如下所示:
storm中bolt与drools的整合 Drools可以将storm中处理数据的规则提取到一个drl文件中,该文件就成了唯一处理规则的文件。任何时候规则出现变化,我们只需要修改该drl文件,而不会改变其它的代码。Bolt与drools的整合代码如下所示:
通过规则处理数据之后,我们就可以将处理过的数据发送到下一个bolt中,然后将数据存储到redis中。
联系客服