基于Spark Streaming的实时数据处理系统设计与实现

摘  要：一般的大数据平台在历史数据处理方面大多都是先通过一些数据导入工具比如Sqoop、DataX等进行数据全量导入，而实时数据处理更加关注的是数据的实时性。针对实时数据处理问题，文章基于Spark Streaming设计实现了一种实时数据处理系统，能够实现高效的实时数据接入、传输、计算校验和存储。该系统具有实时数据获取、实时数据计算、实时数据存储等特点，为进一步从实时数据中获取有效信息提供了必要的基础支撑。

关键词：实时;Flume;Kafka;Spark Streaming;HBase

中图分类号：TP274      文献标识码：A 文章编号：2096-4706（2020）20-0010-03

Design and Implementation of Real-time Data Processing System

Based on Spark Streaming

SHI Zhao

（Nanjing University of Posts and Telecommunications，Nanjing  210023，China）

Abstract：In terms of historical data processing，general big data platforms mostly import full data through some data import tools such as Sqoop，DataX，etc.，while real-time data processing pays more attention to the real-time nature of the data. Aiming at the real-time data processing problem，this paper designs and implements a real-time data processing system based on Spark Streaming，which can realize efficient real-time data access，transmission，calculation，verification and storage. The system has the characteristics of real-time data acquisition，real-time data calculation，real-time data storage，etc.，which provides the necessary basic support for further obtaining effective information from real-time data.

Keywords：real-time;Flume;Kafka;Spark Streaming;HBase

0  引  言

隨着大数据技术的发展，人们越来越重视数据中潜藏着的价值。利用大数据技术，我们可以从数据中挖掘其隐藏的价值，为我们的生产、生活和学习提供有力的指导。社会生产和生活中每时每刻在产生新的实时数据，但我们对这些数据的处理还不充分。如何从这些数据中获取更多有效的信息支撑应用中的实时响应，是目前的研究热点，Spark Streaming技术的出现，为我们实现高效实时数据处理提供了技术支撑。本文使用Spark Streaming技术设计了一个实时数据处理系统，该系统是作者在参与上海德拓信息技术股份有限公司南京分公司关于离线数据采集存储系统的开发与维护的经验基础上提出的一种实时数据处理系统。

1  实时数据处理系统的需求

实时数据的产生多种多样，如用户的浏览信息、游客的出行记录信息、顾客的消费信息等，这些信息会被系统记录在数据库或日志中。导入这些数据，需要对这些数据库或者日志监控，数据写入数据库或者日志后通过监控获取这些数据，还需要经过数据的清洗。数据录入错误、数据的数值不正确等多种原因导致的脏数据，若不经过校验和清洗就直接导入到库中，会产生极大的成本和时间代价。因此获取到的数据需要经过数据清洗后存入到数据库以避免脏数据带来的影响。

2  实时数据处理系统分析

一个实时数据处理过程包含了数据的接入、数据的传输、数据的计算校验和数据的存储，其具体流程如图1所示。首先需要有数据接入，有了数据之后需要将数据传输到相应位置等待数据计算校验，经过计算校验之后的数据才能存储进数据库。

2.1  数据接入分析

根据所需获取的数据的来源不同，可分为两种。若是获取数据库中的实时增加的数据，可以开启数据库的binlog日志[1]，然后配置canal中deployer的instances.properties配置文件和adapter中的application.yml以及hbase.yml配置文件，实现源库和目标库之间数据同步，原理就是数据库之间的主从复制。若是获取的数据为日志文件中的新增数据，我们可以使用Flume这样一个高可用的海量日志采集聚合传输的工具[2]。Flume用来监测日志文件的变化，一旦日志文件的内容发生变化，Flume便可以获取新增的数据内容。Flume由Source，Channel和Sink组成，Source负责完成数据的收集，Channel对Source提供的数据进行缓存，Sink取出Channel中的数据，存入到相应的文件系统（HDFS）、数据库或者Kafka中，其系统架构图如图2所示。

2.2  数据传输分析

数据的接入速率与数据的处理速率不同，这就需要在数据接入和数据处理之间加上一个缓冲区。这个缓冲区必须是高性能的，而且可用于实时事件响应的场景，Kafka正满足这些需求。Kafka是Apache下的一个开源流处理平台，可以处理用户在日常生活中的所有动作流数据，同时Kafka可以以集群模式运行，缓解节点或者服务器之间的压力。因此我们可以将由Flume监测日志文件而获取的数据发送给Kafka的topic，存储在Kafka的缓冲区中，等待Kafka消费者将这些采集到的数据“消费”。

2.3  数据计算校验分析

一般的数据计算方法是通过MapReduce完成，但Map-Reduce仅仅支持map和reduce操作，操作单一，map的中间结果写入磁盘，reduce的结果写入HDFS，大数据量的MapReduce操作所花费的时间会很高，因此MapReduce不适合用于实时计算的场景。spark是内存计算，避免了多次计算的中间结果写到HDFS的I/O开销，且spark提供的RDD操作很多[3]。因此Spark Streaming[4]用在数据计算校验部分正合适。

2.4  数据存储分析

实时数据导入系统中将经过Spark Streaming流式处理[5]后的数据存储到HBase中[6]。HBase具有海量存储，高并发，极易扩展，成本较低等特点，而且HBase可以同时存储多版本数据（HBase使用不同Timestamp来标识相同Rowkey行对应的不同版本数据）。HBase查询数据的响应速度很快，这是因为HBase的特殊寻址方式（请求ZooKeeper获取元数据，访问元数据获取RegionServer地址，访问RegionServer获取所需数据），寻址访問的同时会把元数据的相关信息缓存下来。这样的访问方式成就了HBase的快速响应的特点。HBase表中Rowkey的设计至关重要。HBase中有很多region，每个region都有startRowKey和stopRowKey，若Rowkey设计不合理，会导致某个region被频繁访问，造成热点现象，引起节点性能下降。因此HBase的Rowkey设计需要注意Rowkey的长度一般不超过16个字节;需要保证Rowkey的唯一性（HBase表中的数据是以KeyValue的形式存在的，若插入相同Rowkey的值则原先的数据会被覆盖）;Rowkey设计要充分利用其有序性;设计的Rowkey应该均匀分布在各个HBase节点上。

3  实时数据处理系统的设计与实现

通过上一节的分析，本文实现了一种实时数据处理系统，通过Flume实时数据获取模块获取日志文件和数据库中的实时新增数据，再将数据暂存到Kafka消息中间件，由Spark Streaming调用Kafka中的数据做实时处理，处理后的结构存储到HBase中。其结构图如图3所示。

3.1  数据接入设计

数据接入以Flume为例，Flume是由Source、Channel和Sink组成，需要将这三部分配置好并串联起来。设置Source端的监测命令为tail -F/root/flume.log，监测flume.log文件内容的变化，若tail-F/root/flume.log命令监测到日志文件内容变动，Flume会获取这些内容。

3.2  数据传输设计

设置flume sink端的类型为Kafka，配置Kafka的boot-strapserver地址并设置Kafka的topic。除此之外还需要设置序列化方式为kafka.serializer.StringEncoder。通过Kafka的bootstrapserver和对应的topic，才能将Source端收集到的数据准确无误的传输到Kafka指定的topic。配置好Source和Sink后，需要再配置一个Channel将Source端收集到的数据传输到Kafka中。

3.3  数据计算校验设计

数据存储到了Kafka的topic中，需要创建一个消费者消费采集到的数据。这里的消费者是通过Spark Streaming实现的，通过Spark Streaming对采集到的数据进行计算校验。首先需要将Spark Streaming与Kafka连接，才能读取到topic中的数据。因此需要配置bootstrapserver、key.deserializer、value.deserializer，设置topic、groupid，kafkaoffset的维护等级，关闭自动提交。

经过上一步取到topic中的数据后，就需要对数据进行计算校验了。先将数据以空格切分得到一个字符串数组，若数组长度或者数组中的数据值不符合需求则不存储这条数据。以WordCount为例，读取topic中的数据后首先需要将这一条数据以空格分隔存储在数组中，然后遍历这个数组并将其转换成一个元组，最后将相同key值的元组聚合得到单词出现的次数。其中获取到的数据经flatmapRDD、mapToPairRDD和reduceByKeyRDD计算获得最终结果。

3.4  数据存储设计

将Spark Streaming处理计算后的数据存入HBase，需要先在HBase数据库中创建好对应的表，建表语句如下：create ‘wordcount，sz，指定表名为WordCount，列族为sz。这里以WordCount为例，统计每个单词出现的次数，因此Rowkey就是每个RDD中tuple的key值，也就是这些出现过的单词。表建好后，需要将经过RDD处理计算好的数据插入到HBase表中。

首先轮训RDD中所有的元组，然后创建HBase配置对象，指定ZooKeeper的端口号和ZooKeeper的节点，指定HBase的HMaster节点，创建连接对象并指定表名从而与HBase连接并把每个元组的key和value插入HBase表中。其中key为HBase表的Rowkey，value为对应列族sz，对应列num的cell中的值。

当第一批数据处理完后第二批数据处理时，有可能会有相同的key的数据，由于HBase可以存储多版本数据，这样的相同key的数据插入进数据库时，会被HBase当作新版本的数据存储下来，并不能实现累加，这显然与WordCount计数的需求不同。因此需要删除表中原有的这一条数据，插入新值与原值的和。因此需要获取表中的所有Rowkey，每当要插入数据时，先检测是否会有相同的Rowkey出现，若表中不存在相同的Rowkey则直接插入该条数据;若表中存在相同的Rowkey，则需要记录原有的值并删除这条数据，插入原值和现有值的和从而达到计数的目的。

3.5  系统运行介绍

整个系统是检测日志文件的，所以先要向日志文件里写入数据，如图4所示。

運行Spark Streaming代码，读取监测到的数据并做计算处理，程序输出如图5所示。

查看HBase中的数据，检验计算后的数据是否成功存入，HBase中的数据如图6所示。

4  结  论

随着大数据技术的不断发展，企业越来越意识到实时数据中的价值。因此本文设计了一种基于Spark Streaming的实时数据处理系统，能够实现实时数据的接入、传输、计算校验和存储。在互联网时代，大数据技术已经日益成熟，许多企业已经开始着眼于数据的隐藏价值，并开始着手建立基于大数据的分布式数据采集和数据应用平台，该实时数据处理系统可以使企业快人一步处理实时数据。但此系统也有些不足，实时数据处理系统只是将实时数据采集并存储起来，并没有对数据进行应用，后续可用机器学习技术对数据的价值进行深层次的挖掘。

参考文献：

[1] 苏子权.基于MySQL Binlog的数据增量同步系统的设计与实现 [D].南京：南京大学，2018.

[2] 袁昌权，胡益群，许光，等.基于Hadoop的高可用数据采集与存储方案 [J].电子技术与软件工程，2019（18）：169-170.

[3] 吴信东，嵇圣硙.MapReduce与Spark用于大数据分析之比较 [J].软件学报，2018，29（6）：1770-1791.

[4] 柯杰.基于Spark Streaming日志实时监测系统的设计与实现 [D].南京：东南大学，2017.

[5] 李欣.基于Spark/HBase的交通流数据存储及索引模型探讨 [J].地理与地理信息科学，2019，35（4）：1-8.

作者简介：施炤（1995.07—），男，汉族，江苏镇江人，硕士研究生，研究方向：大数据技术。