Flume＋Kafka＋Storm的实时日志统计

在整合这套方案的时候，项目组也是经过一番讨论，在讨论中，观点很多，有人认为直接使用Storm进行实时处理，去掉Kafka环节；也有认为直接使用Kafka的API去消费，去掉Storm的消费环节等等，但是最终组内还是一致决定使用这套方案，原因有如下几点：

• 业务模块化
• 功能组件化
  

我们认为，Kafka在整个环节中充当的职责应该单一，这项目的整个环节她就是一个中间件，下面用一个图来说明这个原因，如下图所示：



整个项目流程如上图所示，这样划分使得各个业务模块化，功能更加的清晰明了。

• Data Collection
  

　　负责从各个节点上实时收集用户上报的日志数据，我们选用的是Apache的Flume NG来实现。

• Data Access
  

　　由于收集的数据的速度和数据处理的速度不一定是一致的，因此，这里添加了一个中间件来做处理，所使用的是Apache的Kafka，关于Kafka集群部署，大家可以参考我写的《Kafka实战－Kafka Cluster》。另外，有一部分数据是流向HDFS分布式文件系统了的，方便于为离线统计业务提供数据源。

• Stream Computing
  

　　在收集到数据后，我们需要对这些数据做实时处理，所选用的是Apache的Storm。关于Storm的集群搭建部署博客后面补上，较为简单。

• Data Output
  

　　在使用Storm对数据做处理后，我们需要将处理后的结果做持久化，由于对响应速度要求较高，这里采用Redis＋MySQL来做持久化。整个项目的流程架构图，如下图所示：



Flume是一个分布式的、高可用的海量日志收集、聚合和传输日志收集系统，支持在日志系统中定制各类数据发送方（如：Kafka，HDFS等），便于收集数据。Flume提供了丰富的日志源收集类型，有：Console、RPC、Text、Tail、Syslog、Exec等数据源的收集，在我们的日志系统中目前我们所使用的是spooldir方式进行日志文件采集，配置内容信息如下所示：

producer.sources.s.type = spooldirproducer.sources.s.spoolDir = /home/hadoop/dir/logdfs当然，Flume的数据发送方类型也是多种类型的，有：Console、Text、HDFS、RPC等，这里我们系统所使用的是Kafka中间件来接收，配置内容如下所示：
producer.sinks.r.type = org.apache.flume.plugins.KafkaSinkproducer.sinks.r.metadata.broker.list=dn1:9092,dn2:9092,dn3:9092producer.sinks.r.partition.key=0producer.sinks.r.partitioner.class=org.apache.flume.plugins.SinglePartitionproducer.sinks.r.serializer.class=kafka.serializer.StringEncoderproducer.sinks.r.request.required.acks=0producer.sinks.r.max.message.size=1000000producer.sinks.r.producer.type=syncproducer.sinks.r.custom.encoding=UTF-8producer.sinks.r.custom.topic.name=test



Kafka是一种提供高吞吐量的分布式发布订阅消息系统，她的特性如下所示：

通过磁盘数据结构提供消息的持久化，这种结构对于即使数据达到TB+级别的消息，存储也能够保持长时间的稳定。
搞吞吐特性使得Kafka即使使用普通的机器硬件，也可以支持每秒数10W的消息。
能够通过Kafka Cluster和Consumer Cluster来Partition消息。
　　Kafka的目的是提供一个发布订阅解决方案，他可以处理Consumer网站中的所有流动数据，在网页浏览，搜索以及用户的一些行为，这些动作是较为关键的因素。这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。对于Hadoop这样的日志数据和离线计算系统，这样的方案是一个解决实时处理较好的一种方案。




Storm的主要特点如下：
简单的编程模型。类似于MapReduce降低了并行批处理复杂性，Storm降低了进行实时处理的复杂性。
可以使用各种编程语言。你可以在Storm之上使用各种编程语言。默认支持Clojure、Java、Ruby和Python。要增加对其他语言的支持，只需实现一个简单的Storm通信协议即可。
容错性。Storm会管理工作进程和节点的故障。
水平扩展。计算是在多个线程、进程和服务器之间并行进行的。
可靠的消息处理。Storm保证每个消息至少能得到一次完整处理。任务失败时，它会负责从消息源重试消息。
快速。系统的设计保证了消息能得到快速的处理，使用ØMQ作为其底层消息队列。
本地模式。Storm有一个本地模式，可以在处理过程中完全模拟Storm集群。这让你可以快速进行开发和单元测试。
　　Storm集群由一个主节点和多个工作节点组成。主节点运行了一个名为“Nimbus”的守护进程，用于分配代码、布置任务及故障检测。每个工作节 点都运行了一个名为“Supervisor”的守护进程，用于监听工作，开始并终止工作进程。Nimbus和Supervisor都能快速失败，而且是无 状态的，这样一来它们就变得十分健壮，两者的协调工作是由Apache的ZooKeeper来完成的。
　　Storm的术语包括Stream、Spout、Bolt、Task、Worker、Stream Grouping和Topology。Stream是被处理的数据。Spout是数据源。Bolt处理数据。Task是运行于Spout或Bolt中的 线程。Worker是运行这些线程的进程。Stream Grouping规定了Bolt接收什么东西作为输入数据。数据可以随机分配（术语为Shuffle），或者根据字段值分配（术语为Fields），或者广播（术语为All），或者总是发给一个Task（术语为Global），也可以不关心该数据（术语为None），或者由自定义逻辑来决定（术语为 Direct）。Topology是由Stream Grouping连接起来的Spout和Bolt节点网络。