基于spark的数据质量评估(大数据巡检系统)

本文主要讲解如何使用基于Spark的公安大数据实时运维技术。感兴趣的朋友不妨看看。本文介绍的方法简单、快速、实用。让边肖带你学习“如何使用基于Spark的公安大数据实时运维技术”！

公安行业有成千上万的前端和后端设备。前端设备包括摄像头、探测器和传感器，后端设备包括各级中央机房的服务器、应用服务器、网络设备和电力系统。设备数量庞大、种类繁多，给公安内部运维管理带来了巨大挑战。传统的通过ICMP/SNMP、Trap/Syslog等工具对设备进行诊断和分析的方法已经不能满足实际需求。由于公安内部运维管理的特殊性，目前ELK等架构的方法也不能满足需求。为了找到合理的解决方案，我们将目光转向开源架构，搭建了适合公安行业的实时运维管理平台。

实时运维平台整体架构

数据采集层：Logstash Flume，负责采集和过滤各种前端和后端硬件设备输出的Snmp Trap和Syslog日志信息以及应用服务器本身在不同场景下生成的系统和业务日志；

数据传输层：采用高吞吐量分布式消息队列Kafka集群，保证聚合日志和消息的可靠传输；

数据处理层：Spark实时Pull Kafka数据，通过Spark Streaming和RDD运算，进行数据流处理和逻辑分析；

数据存储层：实时数据存储在MySQL中，便于实时业务应用和显示；总数据存储在ES和HBase中，用于后续检索和分析；

服务层：基于存储层，后续整体业务应用涵盖APM、网络监控、拓扑、告警、工单、CMDB等。

整个系统涉及的主要开源框架如下：

此外，整个环境基于JDK 8和Scala 2.10.4。公安系统设备有很多种。接下来，以exchange Syslog日志为例，详细介绍日志处理和分析的整体流程。

图1公安实时运维平台总体架构

Flume+Logstash日志收集

Flume是Cloudera贡献的一个分布式、可靠、高可用的海量日志收集系统，支持定制各种Source进行数据收集，并提供简单的数据处理和通过缓存写入Sink的能力。

水槽中，源、槽和汇的配置如下：

在该配置中，通过syslog源配置localhost tcp 5140端口接收网络设备发送的Syslog信息，将事件缓存在内存中，然后通过KafkaSink将日志发送到kafka集群中名为“syslog-kafka”的主题。

来自Elastic公司的Logstash旨在收集、分析和传输各种日志、事件和非结构化数据。它有三个主要功能：事件输入、事件过滤和事件输出，这些功能都设置在带后缀的配置文件中。糖膏剂在本例中，系统日志配置如下：

输入插件用于指定各种数据源。在本例中，Logstash通过udp 514端口接收Syslog信息。

虽然这个例子中不需要配置Filter插件，但是它的功能非常强大，可以进行复杂的逻辑处理，包括正则表达式处理、编解码、k/v分割以及数值、时间等各种数据处理，可以根据实际场景进行设置。

输出插件用于将处理后的事件数据发送到指定的目的地，并指定Kafka的位置、主题和压缩类型。在* * * *的Codec插件中，将源主机的IP地址(主机)和Logstash处理的时间戳(@timestamp)指定为前缀，并集成原始事件消息，便于在事件传输过程中判断Syslog信息的来源。原始Syslog信息流的示例如下：

14712164: 10月9日18:04:10.7353360% LINK-3-updown :接口千兆以太网0/16，已将状态更改为关闭

日志输出插件处理的信息流变成：

19.1.1.12 2016-10-13t 10:04:5

4.520Z <147>12164: Oct 9 18:04:10.735:  %LINK-3-UPDOWN: Interface GigabitEthernet0/16, changed state to down

其中红色字段就是codec编码插件植入的host以及timestamp信息。处理后的Syslog信息会发送至Kafka集群中进行消息的缓存。

Kafka日志缓冲

Kafka是一个高吞吐的分布式消息队列，也是一个订阅/发布系统。Kafka集群中每个节点都有一个被称为broker的实例，负责缓存数据。Kafka有两类客户端，Producer(消息生产者的)和Consumer(消息消费者)。Kafka中不同业务系统的消息可通过topic进行区分，每个消息都会被分区，用以分担消息读写负载，每个分区又可以有多个副本来防止数据丢失。消费者在具体消费某个topic消息时，指定起始偏移量。Kafka通过Zero-Copy、Exactly  Once等技术语义保证了消息传输的实时、高效、可靠以及容错性。

Kafka集群中某个broker的配置文件server.properties的部分配置如下：

其中需指定集群里不同broker的id，此台broker的id为1，默认监听9092端口，然后配置Zookeeper(后续简称zk)集群，再启动broker即可。

Kafka集群名为syslog-kafka的topic：

Kafka集群的topic以及partition等信息也可以通过登录zk来观察。然后再通过下列命令查看Kafka接收到的所有交换机日志信息：

部分日志样例如下：

Spark日志处理逻辑

Spark是一个为大规模数据处理而生的快速、通用的引擎，在速度、效率及通用性上表现极为优异。

在Spark主程序中，通过Scala的正则表达式解析Kafka Source中名为“syslog-kafka”  的topic中的所有Syslog信息，再将解析后的有效字段封装为结果对象，***通过MyBatis近实时地写入MySQL中，供前端应用进行实时地可视化展示。另外，全量数据存储进入HBase及ES中，为后续海量日志的检索分析及其它更高级的应用提供支持。主程序示例代码如下：

整体的处理分析主要分为4步：

初始化SparkContext并指定Application的参数;

创建基于Kafka topic “syslog-kafka” 的DirectStream;

将获取的每一行数据映射为Syslog对象，调用Service进行对象封装并返回;

遍历RDD，记录不为空时保存或者更新Syslog信息到MySQL中。

Syslog POJO的部分基本属性如下：

SwSyslog实体中的基本属性对应Syslog中的接口信息，注解中的name对应MySQL中的表sw_syslog  以及各个字段，MyBatis完成成员属性和数据库结构的ORM(对象关系映射)。

程序中的SwSyslogService有两个主要功能：

encapsulateSwSyslog()将Spark处理后的每一行Syslog通过Scala的正则表达式解析为不同的字段，然后封装并返回Syslog对象;遍历RDD分区生成的每一个Syslog对象中都有ip以及接口信息，saveSwSyslog()会据此判断该插入还是更新Syslog信息至数据库。另外，封装好的Syslog对象通过ORM工具MyBatis与MySQL进行互操作。

获取到的每一行Syslog信息如之前所述：

这段信息需解析为设备ip、服务器时间、信息序号、设备时间、Syslog类型、属性、设备接口、接口状态等字段。Scala正则解析逻辑如下：

通过正则过滤、Syslog封装以及MyBatis持久层映射，Syslog接口状态信息最终解析如下：

***，诸如APM、网络监控或者告警等业务应用便可以基于MySQL做可视化展示。

到此，相信大家对“基于Spark的公安大数据实时运维技术怎么使用”有了更深的了解，不妨来实际操作一番吧！这里是网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！