大数据案例互联网日志实时收集和实时计算的简单方案

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

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

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

为了解决数据实时跨外网传输以及实时业务的问题&＃xff0c;于是有了现在的架构&＃xff1a;

1. 引入Kafka&＃xff0c;并且和日志收集服务器部署在北京同机房&＃xff1b;

    

2. 每台日志收集服务器上的Flume Agent&＃xff0c;通过内网将数据发送至Kafka&＃xff1b;

    

3. Kafka的第一个消费者&＃xff0c;北京网关机上的Flume&＃xff0c;负责从Kafka中消费数据&＃xff0c;然后流到北京Hadoop集群&＃xff1b;

    

4. Kafka的第二个消费者&＃xff0c;西安网关机上的Flume&＃xff0c;负责从Kafka中消费数据&＃xff0c;然后流到西安Hadoop集群&＃xff1b;这里是西安的Flume通过外网连接北京Kafka&＃xff0c;主动拉取数据&＃xff0c;如果网络不稳定&＃xff0c;那么当前批次拉取失败&＃xff0c;最多重新拉一次&＃xff0c;数据不会进Flume channel&＃xff0c;更不会流到HDFS上&＃xff0c;因此&＃xff0c;这种方式在网络不稳定的情况下&＃xff0c;不会造成数据缺失或重复&＃xff1b;

    

5. Kafka的第三个消费者&＃xff0c;北京网关机上的实时计算模块&＃xff0c;后面再说&＃xff1b;

    

6. Kafka的第N个消费者&＃xff0c;其他&＃xff1b;

Kafka中的数据分区及副本

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

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

Flume拦截器的使用

在整个流程中&＃xff0c;有两个地方用到了同一个Flume拦截器&＃xff08;Regex Extractor Interceptor&＃xff09;&＃xff0c;就是在Flume Source中从消息中提取数据&＃xff0c;并加入到Header&＃xff0c;供Sink使用&＃xff1b;

1. 一处是在LogServer上部署的Flume Source&＃xff0c;它从原始日志中提取出用户ID&＃xff0c;然后加入到Header中&＃xff0c;Flume Sink&＃xff08;Kafka Sink&＃xff09;再入Kafka之前&＃xff0c;从Header中拿出该用户ID&＃xff0c;然后通过应用分区规则&＃xff0c;将该条消息写入Kafka对应的分区中&＃xff1b;

2. 另外一处是部署在西安的Flume Source&＃xff0c;它从Kafka中读取消息之后&＃xff0c;从消息中抽取出时间字段&＃xff0c;并加入到Header中&＃xff0c;后面的Flume Sink&＃xff08;HDFS Sink&＃xff09;通过读取Header中时间&＃xff0c;根据消息中的时间&＃xff0c;将数据写入HDFS相应的目录和文件中。

如果在HDFS Sink中仅仅使用当前时间来确定HDFS目录和文件名称&＃xff0c;这样会造成一小部分数据没有写入到正确的目录和文件中&＃xff0c;比如&＃xff1a;日志中8点59分59秒的数据可能会被写进HDFS上9点的目录和文件中&＃xff0c;因为原始数据经过Kafka&＃xff0c;通过外网传输到西安的Flume&＃xff0c;有个几秒的延时&＃xff0c;那是很正常的。

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

在北京部署的Flume&＃xff0c;使用Kafka Source从Kafka中读取数据流向北京Hadoop集群&＃xff0c;西安的也一样&＃xff0c;在消费同一Topic的消息时候&＃xff0c;我们都是在两台机器上启动了两个Flume Agent&＃xff0c;并且设置的统一消费组&＃xff08;group.id&＃xff09;&＃xff0c;根据Kafka相同的Topic&＃xff0c;一条消息只能被同一消费组内的一个消费者消费&＃xff0c;因此&＃xff0c;Kafka中的一条消息&＃xff0c;只会被这两个Flume Agent其中的一个消费掉&＃xff0c;如果一个Flume Agent挂掉&＃xff0c;那么另外一个将会消费所有消息&＃xff1b;想系统学习大数据的话&＃xff0c;可以加入大数据技术学习扣扣君羊&＃xff1a;522189307

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

实时计算模块

目前我们实时计算的业务比较简单&＃xff0c;就是类似于根据不同维度统计PV和UV。比如&＃xff1a;实时统计一个网站当天的累计PV、UV、IP数等&＃xff0c;目前我们直接开发的JAVA程序&＃xff0c;使用streamlib统计这些指标&＃xff0c;UV和IP数这种需要去重的指标有2%以内的误差&＃xff0c;业务可以接受。

实时计算模块使用Kafka low-level API&＃xff0c;针对每一个Topic&＃xff0c;都使用和分区数相等的线程去处理&＃xff0c;每个线程消费一个分区的数据&＃xff0c;由于数据在进入Kafka分区的时候&＃xff0c;都是经过相应规则的分区&＃xff0c;因此相同用户的数据会在同一个分区中&＃xff1b;

另外&＃xff0c;每个线程会在Redis中维护自己当前的Offsets&＃xff0c;比如&＃xff1a;在实时计算当天累计指标的业务场景中&＃xff0c;每天0天在Redis中记录当前的Offsets&＃xff0c;这样&＃xff0c;如果实时计算程序挂掉&＃xff0c;下次启动时候&＃xff0c;从Redis中读取当天的Offsets&＃xff0c;重新读取和计算当天的所有消息。

由于我们的需求是实时统计当天累计的指标&＃xff0c;而且能接受一定的误差&＃xff0c;因此采用这种方式。如果需要精确统计累计去重指标&＃xff0c;那么可能需要采用其它方式&＃xff0c;比如&＃xff1a;精确统计当天实时累计用户数&＃xff0c;一种简单的办法是在HBase中使用计数器来配合完成。

其它实时数据消费者

如果需要实时统计一小段时间&＃xff08;比如十分钟、一小时&＃xff09;之内的PV、UV等指标&＃xff0c;那么可以使用SparkStreaming来完成&＃xff0c;比较简单。如果单独使用Spark Streaming来完成一天内海量数据的累计去重统计&＃xff0c;我还不太清楚有什么好的解决办法。

另外&＃xff0c;实时OLAP也可能作为Kafka的实时消费者应用&＃xff0c;比如:Druid。