水利自动化实时流式大数据的处理研究

彭大为，史惠存

(1.安徽省水利水电勘测设计研究总院有限公司，230088，合肥；2.江苏省建筑工程质量检测中心有限公司，210028，南京)

0 引言

根据《促进大数据发展行动纲要》和《全国水利信息化发展“十三五”规划》提出加快推动数据资源共享开放和开发应用， 实施国家大数据战略。《关于推进水利大数据发展的指导意见》指出应用大数据处理技术处理多源、海量、动态、持续增加的水利数据，以满足智慧水利的建设要求。在《水利部关于印发加快推进智慧水利的指导意见和智慧水利总体方案的通知》中强调基于物联网建设空天地一体化数据感知网络，采集水利相关数据，为水利大脑提供数据支撑。中国水利企业协会智慧水利分会在智慧水利与河湖长制高峰论坛暨2019智慧水利分会年中再一次围绕云计算、大数据、物联网等技术实现由数字水利向智慧水利转变。水利大数据是实现智慧水利的前提和基础。

随着水利行业智能传感器的发展和大规模的应用以及物联网技术的逐渐成熟，水利行业数据采集能力不断提升，形成水利大数据。水利大数据是由水利业务数据、水利相关行业和领域数据构成。龚琪慧[1]等提出水利大数据具有数据量大、数据来源形式多样、数据持续增长、数据价值高和数据有实时性或准实时要求等特征。陈蓓青[2]等认为水利大数据具有数据量大、数据类型复杂和计算过程复杂耗时等特点。

水利数据分为结构化数据、非结构化、半结构化数据，结构化数据是关系型数据，有数据结构规范；非结构化数据是数据结构不完整不规则的数据形式；半结构化数据通常称为自描述结构数据，是结构化数据的一种，包含相关标记，用来分隔语义元素以及对记录和字段进行分层，如XML , GeoJson数据格式。为了高效处理水利大数据，通过采用分布式数据处理集群组建云环境，对不同结构的数据进行相应的处理。

传统水利数据的处理是基于专业和部门的内部数据进行抽样分析，而水利大数据的处理是对跨部门、跨领域、多维度的海量数据进行整体分析处理。水利大数据分析一般是指利用分布式计算集群对数据进行处理、挖掘和可视化操作等， 主要有批处理和流处理2种方式[3]。MapReduce是目前较为常用的批处理引擎，但是在处理海量高频实时数据时吞吐量较大，不能做到低延迟，不适合实时处理。Spark Streaming是Spark核心API的一个扩展，可以实现高吞吐量的、具备容错机制的实时流数据的处理[4]。

本文主要阐述对水利自动化设备上报的实时流式大数据的处理方法，特别是通过PLC技术实时获取自动化设备的运行参数，该数据上报频率在毫秒级别，数据量大、数据源多、格式不同、数据持续等特点，使得对该类的数据处理一般传统的数据处理模式已经无法满足，因此针对这种数据设计采用Spark Streaming技术进行处理，通过测试已证明本方法能够对水利自动化实时流式大数据进行高效处理。

1 Rocket MQ

RocketMQ是一个由阿里巴巴开源的消息中间件，在设计上借鉴了Kafka。2012年开源，2017年成为apache顶级项目。RoketMQ结构体如图1所示。

图1 RoketMQ结构体

由Producer(生产者)、Topic(消息队列)、Consumer(消费者)3部分组成。消息通过Topic进行传递。Topic存放的是消息的逻辑地址。Producer将消息发往具体的Topic。Consumer订阅Topic，主动拉取或被动接受消息。

Topic是一个逻辑上的概念，每个逻辑队列保存一部分消息数据，但是保存的消息数据实际上不是真正的消息数据，而是指向Commit log的消息索引。Top可采用集群的方式进行创建，也可通过单个Broker模式去创建，创建的每个Topic的角色相同。

Roket MQ为了实现高吞吐量、高并发，通常一个Topic被分配到多个代理中，每个代理包含多个Topic分区，每个Topic分区中保存相同类型的Queue。 Topic Broker Queue关系如图2所示。

图2 Topic Broker Queue关系

消息队列作为RoketMQ高并发系统的核心组件，能够帮助业务系统结构提升开发效率和系统稳定性。

2 Spark Streaming

流式计算作为大数据处理领域的一种主要模型，当前主流的流计算框架由Twitter公司开发的Storm，Yahoo公司开发S4、微软的Timestream以及UCBerkeley AMPLab开发的Spark Streaming等。

Spark Streaming实时流式大数据集群由多个工作节点组成，每个节点运行多个Spark Executor，在Spark Executor上运行相关业务处理程序。在此可以创建Maven应用来处理水利自动化上报的PLC数据，并将应用和相关Jar包进行完整打包，通过Spark-submit命令将Jar包提交到Spark集群节点中。提交应用的集群节点作为该应用的Driver节点，并从Cluster Manager中获取资源，将根据时间窗口将实时流式数据进行分批形成弹性分布式数据集(RDD)，每次RDD的Action会产生一个新Job，每个Job包含多个Task，Cluster Manager机制根据集群资源分配情况动态地将Task分配到Worker Node中的Executor中处理。 Worker各节点最终处理结果汇集到Driver节点进行汇总输出。

Spark Streaming是基于离散数据流Dstream，构建在Spark计算引擎之上的分布式流式计算框架，具有高吞吐量、具备容错机制的实时流数据处理。Dstream是由时间上连续的弹性分布式数据集RDD序列组成，每个RDD包含一定时间间隔内的数据流，是不可变的，可重算的数据集[5]。

图3 Dstream

Spark将大数据切分后放入RDD作为Spark的基本数据结构，在RDD中进行数据处理操作，Spark根据操作调度集群资源进行计算。RDD的操作主要分为Transformation和Action 2种。Transformation操作表示将一个RDD通过一系列操作变成另一个RDD的过程。Transformation操作不会触发真正的计算，仅建立RDD间的DAG有向无环图。Action代表一次计算结束，不再生成新的RDD，并将结果返回给Driver程序。每个Action操作调用SparkContext的RunJob方法向集群提交请求。

通过有向无环图(DAG)的Narrow窄依赖、Shuffle宽依赖实现Spark streaming的容错。此种容错主要是由于RDD中的数据是不可变的分布式弹性数据集，在集群处理过程中若RDD丢失可根据DAG找到其父RDD重新计算得到。

Narrow Dependency即父RDD与子RDD间的分区一对一，Map、Union操作为窄依赖；Shuffle Dependency即父RDD与子RDD间的对应关系不是一对一关系，Reducebykey操作为宽依赖；值得注意的是Join同属宽、窄依赖如图4所示。

窄依赖 窄·宽依赖

Spark Streaming支持从多种数据源获取数据，如kafka、flume、zeroMQ、RoketMQ等。 Spark Streaming通过定义接口从RoketMQ消费者API中读取数据，分配到Spark Streaming集群节点分区中，从而实现对实时流式大数据进行分布式处理[6]。

由于水利自动化PLC数据的高频上报，数据持续到达，数据实时读入和数据规模大特点的实时流式大数据的处理传统的数据处理框架不能满足需求，而Spark Streaming能够很好地对实时流式大数据和离线大数据进行分布式处理[7]。

3 总体架构

基于RoketMQ和Spark Streaming框架设计了一种水利自动化实时流式数据分布式处理平台。总体架构包括数据源、Netty服务器、RoketMQ、SparkStreaming和持久层，具体如图5所示。

数据源是水利自动化设备通过将下位机中的采集和运行数据，通过PLC数据采集软件将采集的数据由Netty服务器调用RoketMQ生产者接口写入RoketMQ集群节点[8]。RoketMQ集群节点对持续实时上报的数据，根据不同告警类型的PLC数据写入相应的Topic中，经过相关业务处理后的数据去向分为3类，第1类是实时告警信息以主动的方式可通过消息系统SMS(如：移动、联通、电信、网通、阿里云、微信、腾讯等)，将告警信息及时发送到相关负责人，使得告警能够及时快速响应；第2类是将实时上报的PLC数据存入Elasticsearch数据库中，由于Elasticsearch具有很强的数据检索能力，系统在数据查询时时效性很高；第3类是将复杂业务处理的数据传递到Spark Streaming进行集群运算，并将处理结果持久化到时序数据库和关系数据库中，该类数据库可作为水利自动化上报的成果数据库，为系统相关业务应用提供支撑。

图5 总统架构

4 Spark Streaming调优

将水利自动化实时流式大数据应用部署在集群之初，出现应用系统运行慢、占用资源多、不稳定等问题，此时需要对集群进行调优才能达到最佳性能。调优是一个具体性很强的事情，不同的应用及场景优化方式也不同，调优并没有统一的标准。

4.1 接收和处理的并行度

在分布式系统中增加接收和处理数据的并行度是提高整个系统性能的关键，为了提高数据接收和处理的并行度需要考虑数据分片的数量。每次Task仅能处理一个Partition,若Partiton个数过小导致每个Partition的数据量太大，导致内存压力过大，Executor的计算能力不能充分利用；若Partiton个数过多，导致分片过多，执行效率低。

4.2 数据序列化

数据序列化在分布式数据处理过程中对集群性能的影响较为显著。本文主要是采用Java的对象输入流框架Object Output Stream Framework进行序列化，通过采用Java.io.Externalizable接口实现更加精细的控制序列化。

4.3 批处理时间间隔设置

水利自动化PLC上报的实时数据的频率能否与集群的处理速度相对保持稳定，直接影响数据处理的稳定性。若上报的PLC数据的频率过高于集群处理速度则将产生数据积压，上报的PLC数据过低于集群处理速度造成算力浪费。合理的批处理时间间隔设置使得数据上报和处理保持一个相对稳定的状态，保证系统运行稳定，否则产生延迟不断增加，系统运行不稳定。

5 结论

水利大数据处理技术是水利科学发展的必然趋势，是大数据研究重要的领域。根据国家水利部2019年发布的智慧水利建设指导意见，要求构建水利一体化感知网络构建水利数据采集平台，该感知平台基于物联网技术获取水利相关数据，为实现智慧水利提供数据支撑。

由于水利数据多源、异构、量大，特别是水利自动化实时上报的流式数据，在实际业务中往往需要对数据进行高频率或超高频率上报，数据不仅量大而且数据持续到达，同时要求处理响应低延迟，因此对水利大数据进行高效处理是实现智慧水利的核心技术。

根据实验测试证明构建该分布式集群和采用实时流式大数据处理技术能够实现对水利自动化实时流式大数据进行高效处理。在该集群运行之初出现集群的数据处理效率很低、CUP和内存的占用率较高的问题，通过对集群进行多次调优后数据处理效率有较大提升，达到实验设计目标。该水利自动化流式大数据处理可能存在一些不足，需要在项目实践中进一步的改造、扩展、优化以满足实际项目建设需要。