基于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。公安系统设备种类繁多，接下来将以交换机Syslog日志为例，详细介绍日志处理分析的整体流程。

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

Flume+Logstash日志收集

Flume是Cloudera贡献的一个分布式、可靠及高可用的海量日志采集系统，支持定制各类Source(数据源)用于数据收集，同时提供对数据的简单处理以及通过缓存写入Sink(数据接收端)的能力。

Flume中，Source、Channel及Sink的配置如下：

该配置通过syslog source配置localhost tcp  5140端口来接收网络设备发送的Syslog信息，event缓存在内存中，再通过KafkaSink将日志发送到kafka集群中名为“syslog-kafka”的topic中。

Logstash来自Elastic公司，专为收集、分析和传输各类日志、事件以及非结构化的数据所设计。它有三个主要功能：事件输入(Input)、事件过滤器(Filter)以及事件输出(Output)，在后缀为.conf的配置文件中设置，本例中Syslog配置如下：

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

Filter(过滤器)插件虽然在本例中不需要配置，但它的功能非常强大，可以进行复杂的逻辑处理，包括正则表达式处理、编解码、k/v切分以及各种数值、时间等数据处理，具体可根据实际场景设置;

Output(输出)插件用于将处理后的事件数据发送到指定目的地，指定了Kafka的位置、topic以及压缩类型。在***的Codec编码插件中，指定来源主机的IP地址(host)、Logstash处理的时间戳(@timestamp)作为前缀并整合原始的事件消息(message)，方便在事件传输过程中判断Syslog信息来源。单条原始Syslog信息流样例如下：

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

Logstash Output插件处理后的信息流变成为：

19.1.1.12 2016-10-13T10:04:54.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的公安大数据实时运维技术怎么使用”有了更深的了解，不妨来实际操作一番吧！这里是亿速云网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！