Flume+Kafka+Storm+Redis构建大数据实时处理系统：实时统计网站PV、UV+展示

1 大数据处理的常用方法

前面给出的那篇文章是基于MapReduce的离线数据分析案例，其通过对网站产生的用户访问日志进行处理并分析出该网站在某天的PV、UV等数据，对应上面的图示，其走的就是离线处理的数据处理方式，而这里即将要介绍的是另外一条路线的数据处理方式，即基于Storm的在线处理，在下面给出的完整案例中，我们将会完成下面的几项工作：

1.如何一步步构建我们的实时处理系统（Flume+Kafka+Storm+Redis）

2.实时处理网站的用户访问日志，并统计出该网站的PV、UV

3.将实时分析出的PV、UV动态地展示在我们的前面页面上

如果你对上面提及的大数据组件已经有所认识，或者对如何构建大数据实时处理系统感兴趣，那么就可以尽情阅读下面的内容了。

需要注意的是，核心在于如何构建实时处理系统，而这里给出的案例是实时统计某个网站的PV、UV，在实际中，基于每个人的工作环境不同，业务不同，因此业务系统的复杂度也不尽相同，相对来说，这里统计PV、UV的业务是比较简单的，但也足够让我们对大数据实时处理系统有一个基本的、清晰的了解与认识，是的，它不再那么神秘了。

2 实时处理系统架构

我们的实时处理系统整体架构如下：

即从上面的架构中我们可以看出，其由下面的几部分构成：

Flume集群

Kafka集群

Storm集群

从构建实时处理系统的角度出发，我们需要做的是，如何让数据在各个不同的集群系统之间打通（从上面的图示中也能很好地说明这一点），即需要做各个系统之前的整合，包括Flume与Kafka的整合，Kafka与Storm的整合。当然，各个环境是否使用集群，依个人的实际需要而定，在我们的环境中，Flume、Kafka、Storm都使用集群。

3 Flume+Kafka整合

3.1 整合思路

对于Flume而言，关键在于如何采集数据，并且将其发送到Kafka上，并且由于我们这里了使用Flume集群的方式，Flume集群的配置也是十分关键的。而对于Kafka，关键就是如何接收来自Flume的数据。从整体上讲，逻辑应该是比较简单的，即可以在Kafka中创建一个用于我们实时处理系统的topic，然后Flume将其采集到的数据发送到该topic上即可。

3.2 整合过程：Flume集群配置与Kafka Topic创建

3.2.1 Flume集群配置

在我们的场景中，两个Flume Agent分别部署在两台Web服务器上，用来采集Web服务器上的日志数据，然后其数据的下沉方式都为发送到另外一个Flume Agent上，所以这里我们需要配置三个Flume Agent.

3.2.1.1 Flume Agent01

该Flume Agent部署在一台Web服务器上，用来采集产生的Web日志，然后发送到Flume Consolidation Agent上，创建一个新的配置文件flume-sink-avro.conf，其配置内容如下：

配置完成后，启动Flume Agent，即可对日志文件进行监听：

$ flume-ng agent --conf conf -n a1 -f app/flume/conf/flume-sink-avro.conf /dev/null

3.2.1.2 Flume Agent02

配置完成后，启动Flume Agent，即可对日志文件进行监听：

$ flume-ng agent --conf conf -n a1 -f app/flume/conf/flume-sink-avro.conf /dev/null

3.2.1.3 Flume Consolidation Agent

该Flume Agent用于接收其它两个Agent发送过来的数据，然后将其发送到Kafka上，创建一个新的配置文件flume-source_avro-sink_kafka.conf，配置内容如下：

配置完成后，启动Flume Agent，即可对avro的数据进行监听：

$ flume-ng agent --conf conf -n a1 -f app/flume/conf/flume-source_avro-sink_kafka.conf /dev/null

3.2.2 Kafka配置

在我们的Kafka中，先创建一个topic，用于后面接收Flume采集过来的数据：

kafka-topics.sh --create --topic f-k-s --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181 --partitions --replication-factor

3.3 整合验证

启动Kafka的消费脚本：

$ kafka-console-consumer.sh --topic f-k-s --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181

如果在Web服务器上有新增的日志数据，就会被我们的Flume程序监听到，并且最终会传输到到Kafka的f-k-stopic中，这里作为验证，我们上面启动的是一个kafka终端消费的脚本，这时会在终端中看到数据的输出：

$ kafka-console-consumer.sh --topic f-k-s --zookeeper uplooking01:2181,uplooking02:2181,uplooking03:2181 .9.6 0f57c8f5-13e2-428d-ab39-9e87f6e85417 GET /index HTTP/1.1 null null Mozilla/5.0 Windows U Windows NT Gecko/2008070208 Firefox/3.0.1 .55.244 fb953d87-d166-4cb4-8a64-de7ddde9054c GET /check/detail HTTP/1.1 null null Mozilla/5.0 Windows NT WOW64 Trident/7.0 rv:11.0 like Gecko .248.22 9d7bb7c2-00bf-4102-9c8c-3d49b18d1b48 GET /user/add HTTP/1.1 null null Mozilla/4.0 compatible MSIE Windows NT6.0 .249.96 null POST /updateById?id HTTP/1.1 null null Mozilla/5.0 Windows NT WOW64 Trident/7.0 rv:11.0 like Gecko .11.101 aa7f62b3-a6a1-44ef-81f5-5e71b5c61368 GET /getDataById HTTP/1.0 /check/init Mozilla/5.0 Windows U Windows NT Gecko/20070803 Firefox/1.5.0.12

这样的话，我们的整合就没有问题，当然kafka中的数据应该是由我们的storm来进行消费的，这里只是作为整合的一个测试，下面就会来做kafka+storm的整合。

4 Kafka+Storm整合

Kafka和Storm的整合其实在Storm的官网上也有非常详细清晰的文档：http://storm.apache.org/releases/1.0.6/storm-kafka.html，想对其有更多了解的同学可以参考一下。

4.1 整合思路

在这次的大数据实时处理系统的构建中，Kafka相当于是作为消息队列（或者说是消息中间件）的角色，其产生的消息需要有消费者去消费，所以Kafka与Storm的整合，关键在于我们的Storm如何去消费Kafka消息topic中的消息（kafka消息topic中的消息正是由Flume采集而来，现在我们需要在Storm中对其进行消费）。

在Storm中，topology是非常关键的概念。

对比MapReduce，在MapReduce中，我们提交的作业称为一个job，在一个Job中，又包含若干个Mapper和Reducer，正是在Mapper和Reducer中有我们对数据的处理逻辑：

在Storm中，我们提交的一个作业称为topology，其又包含了spout和bolt，在Storm中，对数据的处理逻辑正是在spout和bolt中体现：

即在spout中，正是我们数据的来源，又因为其实时的特性，所以可以把它比作一个“水龙头”，表示其源源不断地产生数据：

所以，问题的关键是spout如何去获取来自kafka的数据？

好在，storm-kafka的整合库中提供了这样的API来供我们进行操作。

4.2 整合过程：KafkaSpout的应用

在代码的逻辑中只需要创建一个由storm-kafkaAPI提供的KafkaSpout对象即可：

spoutConf hosts topic zkRoot id spoutConf

下面给出完整的整合代码：

$ flume-ng agent --conf conf -n a1 -f app/flume/conf/flume-sink-avro.conf /dev/null

其实代码的逻辑非常简单，我们只创建了一个由storm-kafka提供的KafkaSpout对象和一个包含我们处理逻辑的MyKafkaBolt对象，MyKafkaBolt的逻辑也很简单，就是把kafka的消息打印到控制台上。

需要注意的是，后面我们分析网站PV、UV的工作，正是在上面这部分简单的代码中完成的，所以其是非常重要的基础。

4.3 整合验证

上面的整合代码，可以在本地环境中运行，也可以将其打包成jar包上传到我们的Storm集群中并提交业务来运行。如果Web服务器能够产生日志，并且前面Flume+Kafka的整合也没有问题的话，将会有下面的效果。

如果是在本地环境中运行上面的代码，那么可以在控制台中看到日志数据的输出：

$ flume-ng agent --conf conf -n a1 -f app/flume/conf/flume-sink-avro.conf /dev/null

如果是在Storm集群中提交的作业运行，那么也可以在Storm的日志中看到Web服务器产生的日志数据：

这样的话就完成了Kafka+Storm的整合。

5 Storm+Redis整合

5.1 整合思路

其实所谓Storm和Redis的整合，指的是在我们的实时处理系统中的数据的落地方式，即在Storm中包含了我们处理数据的逻辑，而数据处理完毕后，产生的数据处理结果该保存到什么地方呢？显然就有很多种方式了，关系型数据库、NoSQL、HDFS、HBase等，这应该取决于具体的业务和数据量，在这里，我们使用Redis来进行最后分析数据的存储。

所以实际上做这一步的整合，其实就是开始写我们的业务处理代码了，因为通过前面Flume-Kafka-Storm的整合，已经打通了整个数据的流通路径，接下来关键要做的是，在Storm中，如何处理我们的数据并保存到Redis中。

而在Storm中，spout已经不需要我们来写了（由storm-kafka的API提供了KafkaSpout对象），所以问题就变成，如何根据业务编写分析处理数据的bolt。

5.2 整合过程：编写Storm业务处理Bolt

5.2.1 日志分析

我们实时获取的日志格式如下：

$ flume-ng agent --conf conf -n a1 -f app/flume/conf/flume-sink-avro.conf /dev/null

其中需要说明的是第二个字段和第三个字段，因为它对我们统计pv和uv非常有帮助，它们分别是ip字段和mid字段，说明如下：

$ flume-ng agent --conf conf -n a1 -f app/flume/conf/flume-sink-avro.conf /dev/null

因此，根据IP地址，我们可以通过查询得到其所在的省份，并且创建一个属于该省份的变量，用于记录pv数，每来一条属于该省份的日志记录，则该省份的pv就加1，以此来完成pv的统计。

而对于mid，我们则可以创建属于该省的一个set集合，每来一条属于该省份的日志记录，则可以将该mid添加到set集合中，因为set集合存放的是不重复的数据，这样就可以帮我们自动过滤掉重复的mid，根据set集合的大小，就可以统计出uv。

在我们storm的业务处理代码中，我们需要编写两个bolt：

第一个bolt用来对数据进行预处理，也就是提取我们需要的ip和mid，并且根据IP查询得到省份信息；

第二个bolt用来统计pv、uv，并定时将pv、uv数据写入到Redis中；

当然上面只是说明了整体的思路，实际上还有很多需要注意的细节问题和技巧问题，这都在我们的代码中进行体现，我在后面写的代码中都加了非常详细的注释进行说明。

5.2.2 编写第一个Bolt：ConvertIPBolt

根据上面的分析，编写用于数据预处理的bolt，代码如下：

$ flume-ng agent --conf conf -n a1 -f app/flume/conf/flume-sink-avro.conf /dev/null

5.2.3 编写第二个Bolt：StatisticBolt

这个bolt包含我们统计网站pv、uv的代码逻辑，因此非常重要，其代码如下：

$ flume-ng agent --conf conf -n a1 -f app/flume/conf/flume-sink-avro.conf /dev/null

5.2.4 编写Topology

我们需要编写一个topology用来组织前面编写的Bolt，代码如下：

$ flume-ng agent --conf conf -n a1 -f app/flume/conf/flume-sink-avro.conf /dev/null

5.3 整合验证

将上面的程序打包成jar包，并上传到我们的集群提交业务后，如果前面的整合没有问题，并且Web服务也有Web日志产生，那么一段时间后，我们就可以在Redis数据库中看到数据的最终处理结果，即各个省份的uv和pv信息：

需要说明的是mid信息是一个set集合，只要求出该set集合的大小，也就可以求出uv值。

至此，准确来说，我们的统计pv、uv的大数据实时处理系统是构建完成了，处理的数据结果的用途根据不同的业务需求而不同，但是对于网站的pv、uv数据来说，是非常适合用作可视化处理的，即用网页动态将数据展示出来，我们下一步正是要构建一个简单的Web应用将pv、uv数据动态展示出来。

6 数据可视化处理

数据可视化处理目前我们需要完成两部分的工作：

1.开发一个Web项目，能够查询Redis中的数据，同时提供访问的页面

2.自行开发或找一个符合我们需求的前端UI，将Web项目中查询到的数据展示出来

对于Web项目的开发，因个人的技术栈能力而异，选择的语言和技术也有所不同，只要能够达到我们最终数据可视化的目的，其实都行的。这个项目中我们要展示的是pv和uv数据，难度不大，因此可以选择Java Web，如Servlet、SpringMVC等，或者Python Web，如Flask、Django等，Flask我个人非常喜欢，因为开发非常快，但因为前面一直用的是Java，因此这里我还是选择使用SpringMVC来完成。

至于UI这一块，我前端能力一般，普通的开发没有问题，但是要做出像上面这种地图类型的UI界面来展示数据的话，确实有点无能为力。好在现在第三方的UI框架比较多，对于图表类展示的，比如就有highcharts和echarts，其中echarts是百度开源的，有丰富的中文文档，非常容易上手，所以这里我选择使用echarts来作为UI，并且其刚好就有能够满足我们需求的地图类的UI组件。

因为难度不大，具体的开发流程的这里就不提及了，有兴趣的同学可以直接参考后面我提供的源代码，这里我们就直接来看一下效果好了。