实时海量日志收集处理系统的设计

刘永涛 院志超

摘 要：在校园网用户爆炸式增长的今天，校园网运维任务越来越重。本文结合运维实际，提出了综合运用Apache Flume、Apache Kafka、Spark Streaming等大数据技术来实时处理海量的日志数据信息，并详细介绍了日志数据信息的收集、传输和处理过程。本系统可以实时、高效的收集来自于不同服务器的日志信息，并使用Spark Streaming和Spark core进行日志数据信息的实时在线和离线处理。

关键词：Apache Flume Apache Kafka Spark Streaming Spark core 日志分析

中图分类号：TP311 文献标识码：A 文章编号：1003-9082（2018）08-000-01

随着校园网用户量的爆发式增长，我校网络出口带宽也由原先的1GB升级为13.8GB，暂时满足了校园网用户的上网需求。然而，用户量的暴增，不仅仅导致了校园网出口的阻塞，更是暴露出了各種各样的网络问题。其中，校园网业务服务器的安全问题尤为突出，为了能实时预警其面临的各种问题，并对设备的故障进行分析和定位，需要对日志进行实时收集、分析和处理。通过不断的探索和实践，得出了使用分布式流数据处理架构Spark streaming来处理海量、结构复杂多变的日志数据信息的结论。本文实现了一个以分布式文件系统HDFS为基本的存储系统、分布式日志收集系统Apache Flume为日志收集部件、Kafka为消息处理缓冲区、Spark streaming离散数据流处理框架为核心的海量日志信息处理系统，完全满足了我校服务器运维中对日志数据信息实时收集、处理和分析的要求。

一、日志收集系统的架构

本日志数据信息收集处理系统的结构如图1.1所示。

各种服务器系统所产生的日志数据信息通过指定的端口如514发送到指定的日志收集Apache Flume NG服务器中。在日志数据信息被发送前，使用rsyslog服务将日志数据信息进行规范化、标准化处理，为后续的处理提供统一的数据格式。

将来自于不同操作系统、不同业务系统的日志数据信息由分布式日志收集系统Apache Flume NG直接发送到Hadoop的分布式文件系统HDFS上持久化。

分布式日志收集系统Apache Flume NG将实时接收到的日志数据信息收集起来，发送到消息处理缓冲区Kafka中，为Spark Streaming实时的处理和分析日志数据信息提供数据源。被发送到消息处理缓冲区Kafka中的日志数据信息被实时的处理消费，Spark Streaming将实时接收、处理和分析的最终结果存入非关系型数据库HBASE中，供用户查询和调用。

二、日志数据信息的实时采集

1.标准化

在对日志数据信息进行处理、分析之前，需要对其进行标准化处理。本系统中，将使用Rsyslog格式来标准化。Rsyslog日志收集转发程序可以支持各种各样的数据库，比如Oracle、Mysql等等。在各台服务器上以及分布式日志收集系统Apache Flume NG使用安装完成Rsyslog后，需要对rsyslog.conf配置文件进行配置。

2.日志传输

在海量日志信息处理系统中，各台服务器运行日志的采集是由分布式日志收集系统Apache Flume NG负责的。分布式日志收集系统Apache Flume NG在一台服务器设备上有且仅有一个用来提供日志数据信息的采集、缓冲和发送进程Agent，该Agent由数据源、通道和数据发送池构成。Apache Flume NG将采集到的日志数据信息发送到数据缓冲区中，数据缓冲区将接收到的数据信息发送到数据发送池，数据发送池有HDFS数据发送池、Avro数据发送池等。

在本系统中，数据源一个Agent负责收集一台服务器中的日志数据信息，数据信息以Avro方式传输，最后数据信息发送到数据发送池中，其中数据发送池分为HDFS和Kafka两个。HDFS数据发送池将传输来的数据信息发送到分布式文件系统HDFS中，Kafka数据发送池将传输来的数据信息发送到消息处理缓冲区Kafka中。在这里，为了提升系统的处理性能，将数据缓冲区定义为内存方式。

3.消息处理缓冲区Kafka

消息处理缓冲区Kafka将接收到的日志数据信息交由Spark streaming来实施分析处理，因此，消息处理缓冲区Kafka在这里平衡了日志数据信息的生产和消费过程，提升了应用程序的可用性。

消息处理缓冲区Kafka是一种分布式的消息系统，基于消费者和生产者。其借助高吞吐率可以在集群中实现百万级的数据传输率，从而实现TB级别的数据传输能力。Kafka多个实例构成了Kafka服务器集群，每一个Kafka实例称为Broker，通过使用一个叫做Topic的名字来区分不同的类别。

三、日志数据信息的处理

日志收集处理系统的核心处理功能就是使用Spark来实现在线和离线两种处理方式，在线数据信息的处理使用Spark streaming来进行，离线数据信息的处理则使用Spark core来进行。在线数据处理实现对数据信息的增量型更新，即进行实时计算。离线数据处理则使用Spark core提供的功能进行统计分析。

Spark是一种分布式计算框架，该框架弥补了Hadoop分布式并行计算的缺陷，提供了一种基于Spark core的快速开发分布式应用程序的能力。在本系统中，采用3台2U的高性能服务器作为分布式计算的环境。

1.在线数据处理

Spark Streaming从上下文SparkContext实现程序入口，SparkContext在定义时，使用诸如系统运行模式、应用程序名称等参数进行初始化和实例化。初始化后，使用方法creatDirectStream来创建用来读取指定Topic的Kafka实例。在逻辑处理开始时，需要使用正则表达式对数据进行过滤。日志数据的过滤可以分为两个阶段，第一个处理阶段可以是根据定义好的syslog日志格式来对数据进行粗过滤，第二个处理阶段则对日志部分进行过滤，从而得到符合要求的数据信息。

2.离线数据处理

在离线数据处理部分，则使用Spark core提供的功能函数进行处理，处理的业务逻辑和在线方式一样，同样需要进行应用程序的初始化、日志数据信息的正则表达式匹配和过滤等逻辑处理。

结语

本文设计了一个使用Spark来处理在线、离线日志数据信息收集处理系统，具备良好的水平扩张性，可以有效的提高系统运维人员处理系统故障的效率，具有较高的实用性和可用性。