基于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的公安大數據實時運維技術怎么使用”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！