基于Spark Streaming的实时日志处理平台设计与实现*

薛瑞，朱晓民

（1 北京邮电大学网络与交换技术国家重点实验室，北京 100876；2 东信北邮信息技术有限公司，北京 100191）

1 引言

目前大多数日志分析工具都只能用于单机的数据分析处理，而面对海量化日志处理以及对分析结果低延时反馈的实际需求，传统的日志处理方式已经远远无法满足人们的需要。

Hadoop是Apache软件基金会（Apache Software Foundation）组织下的一个开源项目，提供分布式计算环境下的可靠、可扩展软件，可以方便地在普通硬件上架设自己的大规模机群系统。由于Hadoop优势突出，基于Hadoop的应用已经被大量开发，尤其是在互联网领域，它可以承担PB级别的数据处理，其原生的Mapreduce编程模型也已经成为主流的分布式编程模型。但是，对于在线/近线系统来说，处理在线系统产生的日志，需要低延时高可靠的处理海量数据，Mapreduce编程模型作为离线计算的代表，无法满足日志实时监控、统计分析的需求。

本文介绍了一种基于Spark Streaming的分布式大规模日志实时处理平台的设计方案，其特色主要体现在海量日志存储及处理和实时统计分析两个方面。文章首先阐述了平台的构建方案，之后介绍了系统的各个模块的功能，对企业海量实时日志处理平台建设具有一定的指导意义。

2 设计目标

提供方便快捷的海量日志数据实时处理与统计服务，具体的设计目标如下：（1）提供海量日志数据的安全存储、快速查询服务；（2）提供实时或准实时的日志处理服务；（3）根据不同用户的需求，允许用户添加指定的日志源，并支持基本的数据统计与展示；（4）为了更好的用户体验，通过Web端实现对日志数据的管理和处理统计。

日志信息对运维人员和数据分析人员非常重要。通过日志信息，可以获取软件访问数据、软件系统内部的核心链路和系统异常等。本文提供了一个日志处理平台设计方案，既适用于线上服务的监控，也适用于有效信息的挖掘，具有一定的实际意义。

3 系统设计

3.1 系统设计图

系统设计如图1所示，图中，箭头的方向代表数据的流向：Kafka作为消息队列服务器，不断地接收Web服务器Push的日志数据；同时，数据处理层和数据资源层不断从日志数据获取层拉取日志信息；数据处理后的结果存入HBase数据库，业务逻辑层通过操作HBase数据库获取所需信息，而显示层与逻辑层之间通过API调用进行通信。

3.1.1 日志数据获取层

日志数据获取层的核心是消息队列系统。消息队列系统是消息队列系统通过Kafka实现。Kafka是一个分布式高吞吐量的消息队列系统，Kafka的集群有多个服务器组成，每个类型的消息被定义为Topic，同一Topic内部的消息按照一定的Key和算法被分区存储在不同的服务器上，消息生产者和消费者可以在多个服务器上生产/消费。

本系统可以包含多种日志数据源，如线上系统产生的应用日志、操作系统、安全系统日志等。本系统允许用户通过Web端设置日志数据路径的方式定义日志数据源。日志数据作为消息发布订阅系统的生产者，发布消息时，Kafka客户端先构造一条消息，将消息加入到消息集合中（Kafka支持批量发布，可以往消息集合中添加多条消息，一次性发布），发送消息时，客户端需指定消息所属的消息Topic。

图1 系统设计图

消费者可以通过如下CreateStream方法从Kafka获取消息流：

defcreateStream（ssc: StreamingContext, zookeeperQurum: String,

groupId: String,topic: Map[String, Int]) : ReceiverInputDStream[（String, String)]

本系统中，数据处理层的计算模块作为消费者不断从Kafka服务器拉取消息；同时，数据经过一定的过滤策略存储到HBase中。

3.1.2 数据资源层

考虑到日志数据的快速增长以及快速查询的需求，采用HBase[1]作为存储。HBase是Hadoop的数据库，位于结构化存储层，Hadoop HDFS为HBase提供了高可靠性的底层存储支持，能够支持海量数据的分布式实时存储；同时，HBase会对存储的数据自动切分，并支持高并发读写操作，使海量数据存储具有更强的扩展性；并且，HBase是面向列的KeyValue数据库，可以存储异构数据并且方便范围查询。这些特性刚好解决了日志信息存储模块存在的问题。

在本系统中，同一日志源信息将会存储在一张HBase表中。表的Rowkey设计为（Long.MaxValue（)-requestTime) +requestPath，通过（Long.MaxValue（)-requestTime）可以将最新的日志放在最前面，便于查看，查询某个时间段的记录时，只要设置scan.setStartRow （beginTime)和scan.setStopRow（endTime)，大大提高了scan的效率[2]。另外，这样设计可以保证批量获取的日志信息集中在同一个Region下，在遍历结果时，可以获得更好的性能。同时，HBase数据库也保存数据处理的结果信息，以供查询统计。

3.1.3 数据处理层

数据处理层是本系统的核心。数据处理层需要从日志获取层拉取信息，并实时处理。本系统的数据处理功能基于Spark Streaming实现[3,4]。Spark Streaming是建立在Spark上的实时计算框架，它将流式计算分解成一系列短小的批处理作业，交由Spark引擎处理，每一段数据都转换成Spark中的RDD，然后将Spark Streaming中对DStream的操作变为针对Spark中对RDD的Transformation操作[5]，将RDD经过操作变成中间结果保存在内存中。整个流式计算根据业务的需求可以对中间的结果进行叠加，或者直接存储到HBase。

Spark Streaming的基本原理是将输入数据流以时间片（秒级）为单位进行拆分，然后以类似批处理的方式处理每个时间片数据，其基本原理如图2所示。

图2 Spark Streaming基本原理图

首先，Spark Streaming把实时输入的日志数据流以时间片Δt （如1s）为单位切分成块。Spark Streaming会把每块数据作为一个RDD，并使用RDD操作处理每一小块数据。每个块都会生成一个Spark Job进行处理，最终结果汇总后存入HBase。

3.1.4 业务逻辑层

业务逻辑层处于数据资源层与显示层中间，起到了数据交换中承上启下的作用。本系统中，消息源配置、日志查找、资源配置、报表显示等业务逻辑的实现均在业务逻辑层完成。同时，业务逻辑层还实现了系统的用户管理和用户权限体系逻辑：不同用户对不同消息源的访问权限、进行配置是否有视图权限、编辑权限等。

3.1.5 显示层

显示层位于最外层，为用户提供Web交互式操作的界面，用于显示数据和接收用户输入的数据。本系统中，用户可以通过Web界面添加日志源、查询筛选日志信息、查询日志报表等，管理员也可以通过Web端对不同用户的操作权限进行设定。

3.2 平台核心功能模块分析

3.2.1 日志查询模块

日志查询模块可以根据关键字和日志产生的时间进行查询，底层通过HBase的SQL实现。根据关键字查询是指基于特定的语法查询，依据条件筛选符合条件的字段类型或者字段内容的日志信息，适合查询是否存在业务异常、程序执行的关键逻辑、IO操作等；根据时间查询包括相对时间查询（相对于当前时间点之前5 min、15 min、30 min、1 h、2 h、3 h、1天、1个 月、1年和所有记录）和绝对时间查询（定义筛选的开始时间和结束时间，格式为：yyyy-MM-ddHH:mm:ss，查询该时间段内的所有日志记录）。同时，也可以支持将两种查询方式结合的联合查询。

3.2.2 日志统计分析模块

数据处理层提供了一些基本日志数据统计的接口。目前， 基本的统计功能包括：总数、最大值、最小值、均值、百分位和TopN。比如，提供了计算WebServer日志中某一状态码数量的getStatusCode方法：

其中Option[AccessLogRecord]是分析器的返回值[6]。调用log.filter（line≥getStatusCode（p.parseRecord （line)) == "404").count将返回状态码是404的行数。

3.2.3 日志报表视图（Dashboard）模块

Dashboard是将各个显示的报表以组件的形式加入到显示仪表盘中。创建Dashboard输入标题、描述信息即创建完成。一个Dashboard可以添加多个组件，DashBorad创建列表中查看当前Dashboard关联组件的总和、创建人、创建时间。实现编辑、删除等操作。

视图模块可以将查询的日志信息结果默认以报表柱状图的方式查看。目前支持以年、月、周、天、小时、分钟6种时间维度查看报表。报表可以显示当前时间维度日志信息的总数，同时可根据该日志流中选择字段（支持所有字段、自选字段、默认字段）的方式查看相应的统计值。报表支持多种查看类型，包括饼图、条形图折线图、散点图；支持年、月、周、天、小时、分钟等视图方式。

4 结束语

目前，海量日志数据的实时处理成为越来越普遍的需求，但是目前的解决方案均无法同时完美解决海量、实时这两个需求。本文介绍海量日志实时处理方案，基于Hadoop yarn平台，具有低成本、高效率、易于扩展的优点，通过Spark Streaming流数据处理技术实时处理日志信息，同时，使用HBase作为数据存储数据库，很好地满足了海量数据存储、快速查询的需求。为用户提供Web端的可视化界面，更方便用户进行统计查询等操作。然而，本系统中提供的日志计算处理的方式比较单一，只支持部分基本信息的统计分析，比如总数、最小值、最大值、均值、百分位、TopN等，暂不支持复杂的日志信息统计，未来的工作中会对此进行改进。

[1]卓海艺.基于HBase的海量数据实时查询系统设计与实现[J].北京邮电大学, 2013（2):4.

[2]张榆, 马友忠, 孟小峰.一种基于HBase的高效空间关键字查询策略[J].小型微型计算机系统, 2012,33（10):2141-2146.

[3]Liu X, Iftikhar N, Xie X.Survey of real-time processing systems for big data[C]//Proceedings of the 18th International Database Engineering & Applications Symposium.ACM, 2014.

[4]Hunter T, Das T, Zaharia M, et al.Large-scale online expectation maximization with spark streaming[J].2012.

[5]Zaharia M, Das T, Li H, et al.Discretized streams: an efficient and fault-tolerant model for stream processing on large clusters[C]//Proceedings of the 4th USENIX conference on Hot Topics in Cloud Ccomputing.USENIX Association, 2012.

[6]Deng L.Building a Big Data Analytics Service Framework for Mobile Advertising and Marketing[J].