日志实时收集和计算的简单方案

作为互联网公司，网站监测日志当然是数据的最大来源。我们目前的规模也不大，每天的日志量大约1TB。后续90%以上的业务都是需要基于日志来完成，之前，业务中对实时的要求并不高，最多也就是准实时（延迟半小时以上），因此，我们使用Flume将数据收集到HDFS，然后进行清洗和分析。

后来，根据业务需要，我们有了两个Hadoop集群，并且部署在不同的地方（北京和西安），而所有的日志收集服务器在北京，因此需要将日志数据通过外网传输到西安，于是有了这样的部署：

日志实时收集和计算的简单方案

很快，通过Flume流到西安Hadoop集群的数据就遇到了问题，比原始数据多或者少一些，造成这个问题的主要原因是在网络不稳定的情况下，北京 Flume Agent发送到西安Flume Collector的过程中，会发送失败，或者响应失败。另外，之前的数据准实时也不能满足业务的需求。

为了解决数据实时跨外网传输以及实时业务的问题，于是有了现在的架构：

日志实时收集和计算的简单方案

引入Kafka，并且和日志收集服务器部署在北京同机房；
每台日志收集服务器上的Flume Agent，通过内网将数据发送至Kafka；
Kafka的第一个消费者，北京网关机上的Flume，负责从Kafka中消费数据，然后流到北京Hadoop集群；
Kafka的第二个消费者，西安网关机上的Flume，负责从Kafka中消费数据，然后流到西安Hadoop集群；这里是西安的Flume通过外网连接北京Kafka，主动拉取数据，如果网络不稳定，那么当前批次拉取失败，最多重新拉一次，数据不会进Flume channel，更不会流到HDFS上，因此，这种方式在网络不稳定的情况下，不会造成数据缺失或重复；
Kafka的第三个消费者，北京网关机上的实时计算模块，后面再说；
Kafka的第N个消费者，其他；

Kafka中的数据分区及副本

这种架构下，Kafka成为了统一的日志数据提供者，至关重要。我们目前有4台Broker节点，每个Topic在创建时候都指定了4个分区，副本数为2；

数据在进入Kafka分区的时候，使用了Flume的拦截器，从日志中提取用户ID，然后通过HASH取模，将数据流到Kafka相应的分区中。这种方式，一方面，完成了简单的负载均衡，另一方面，确保相同的用户数据都处于同一个分区中，为后面实时计算模块的统计提供了极大的便利。

Flume拦截器的使用

在整个流程中，有两个地方用到了同一个Flume拦截器（Regex Extractor Interceptor），就是在Flume Source中从消息中提取数据，并加入到Header，供Sink使用；

一处是在LogServer上部署的Flume Source，它从原始日志中提取出用户ID，然后加入到Header中，Flume Sink（Kafka Sink）再入Kafka之前，从Header中拿出该用户ID，然后通过应用分区规则，将该条消息写入Kafka对应的分区中；
另外一处是部署在西安的Flume Source，它从Kafka中读取消息之后，从消息中抽取出时间字段，并加入到Header中，后面的Flume Sink（HDFS Sink）通过读取Header中时间，根据消息中的时间，将数据写入HDFS相应的目录和文件中。如果在HDFS Sink中仅仅使用当前时间来确定HDFS目录和文件名称，这样会造成一小部分数据没有写入到正确的目录和文件中，比如：日志中8点59分59秒的数据可能会被写进HDFS上9点的目录和文件中，因为原始数据经过Kafka，通过外网传输到西安的Flume，有个几秒的延时，那是很正常的。

Flume消费者的负载均衡和容错

在北京部署的Flume，使用Kafka Source从Kafka中读取数据流向北京Hadoop集群，西安的也一样，在消费同一Topic的消息时候，我们都是在两台机器上启动了两个 Flume Agent，并且设置的统一消费组（group.id），根据Kafka相同的Topic，一条消息只能被同一消费组内的一个消费者消费，因此，Kafka中的一条消息，只会被这两个Flume Agent其中的一个消费掉，如果一个Flume Agent挂掉，那么另外一个将会消费所有消息；

这种方式，也是在流向HDFS的消费者端做了负载均衡和容错。