顏 冰,王鐘雷

(中國人民財(cái)產(chǎn)保險(xiǎn)股份有限公司 大數(shù)據(jù)中心,北京100022)

大數(shù)據(jù)時(shí)代,社會(huì)的各行各業(yè),人類的衣、食、住、行、醫(yī)、娛等,時(shí)時(shí)都在產(chǎn)生數(shù)據(jù),這些數(shù)據(jù)正呈指數(shù)級(jí)、爆炸式增長,而且隨著信息科技的不斷進(jìn)步,從海量異構(gòu)大數(shù)據(jù)中迅速且高效地挖掘出有效價(jià)值,并將其轉(zhuǎn)化為可靠的決策依據(jù),已經(jīng)成為各個(gè)行業(yè)所面臨的重大的挑戰(zhàn),極大地考驗(yàn)著數(shù)據(jù)統(tǒng)計(jì)分析的能力[1].面對(duì)規(guī)模大、種類多、變化快等大數(shù)據(jù)問題,許多企業(yè)通過大規(guī)模的硬件資源投入來保障數(shù)據(jù)的基本處理能力,從數(shù)據(jù)采集到生成報(bào)表,仍然采用T+1日的數(shù)據(jù)處理機(jī)制,規(guī)模較大的企業(yè)甚至需要8 個(gè)小時(shí)左右才能完成,傳統(tǒng)技術(shù)已不能全面滿足業(yè)務(wù)和管理決策的數(shù)據(jù)時(shí)效要求,企業(yè)指標(biāo)數(shù)據(jù)的實(shí)時(shí)供給能力亟待全面提升,傳統(tǒng)統(tǒng)計(jì)工作到了必須重視、研究和應(yīng)用大數(shù)據(jù)技術(shù)的發(fā)展階段.

為提高數(shù)據(jù)的統(tǒng)計(jì)分析處理能力,很多企業(yè)采用了數(shù)據(jù)一體機(jī)方式作為解決方案,如Teradata 大數(shù)據(jù)一體機(jī).傳統(tǒng)架構(gòu)是主機(jī)、存儲(chǔ)、網(wǎng)絡(luò)、管理軟件、數(shù)據(jù)倉庫(數(shù)據(jù)庫或者中間件或者虛擬化軟件)等進(jìn)行分散管理,而一體機(jī)則是把這些進(jìn)行集成,打包形成一體化的解決方案,來消除傳統(tǒng)解決方案中存在的性能瓶頸問題,比如數(shù)據(jù)管理,I/O 讀寫等方面存在的性能瓶頸,有針對(duì)性地提升系統(tǒng)的整體處理能力.但是隨著數(shù)據(jù)的急劇增長和大數(shù)據(jù)需求的爆發(fā),尤其是單一化場景需求逐步向多元化場景需求的轉(zhuǎn)變,一體機(jī)方式從總體成本、擴(kuò)展能力、配套軟件等方面看,已經(jīng)逐漸失去競爭優(yōu)勢,尤其是對(duì)于大型(數(shù)據(jù)密集型)企業(yè)明顯不是最優(yōu)解決方案.當(dāng)前,越來越多的企業(yè)開始探索并通過應(yīng)用Hadoop 等大數(shù)據(jù)技術(shù)來提升大數(shù)據(jù)治理和實(shí)時(shí)供給能力[2,3].

大數(shù)據(jù)的處理系統(tǒng)大概可以分為兩類,也就是批處理與流處理系統(tǒng)[4].批處理大數(shù)據(jù)系統(tǒng)(以Hadoop為代表)需先將數(shù)據(jù)“匯聚成批”,通過批量的預(yù)處理之后,加載到分析型數(shù)據(jù)倉庫之中,可以用來進(jìn)行高性能離線“實(shí)時(shí)查詢”.這種批處理系統(tǒng)雖然可以對(duì)完整地大數(shù)據(jù)集合實(shí)現(xiàn)高效的即時(shí)查詢,但卻沒有辦法查詢到增量的實(shí)時(shí)在線數(shù)據(jù),存在數(shù)據(jù)延遲的問題.相較于批處理大數(shù)據(jù)系統(tǒng),流處理是一種大數(shù)據(jù)實(shí)時(shí)處理技術(shù)的典型應(yīng)用,它是一個(gè)無限增長、沒有邊界的動(dòng)態(tài)數(shù)據(jù)集合,以Spark Streaming、Storm、Flink 為代表的流處理系統(tǒng)無需存儲(chǔ)大數(shù)據(jù),可對(duì)數(shù)據(jù)進(jìn)行實(shí)時(shí)高效地在線預(yù)處理,全面、逐條地加載到高性能的內(nèi)存數(shù)據(jù)庫中供查詢.本文研究的是一種使用數(shù)據(jù)庫日志分析、流處理、內(nèi)存計(jì)算和分布式等技術(shù)的指標(biāo)實(shí)時(shí)計(jì)算模式,即基于流處理技術(shù)的指標(biāo)實(shí)時(shí)計(jì)算方法.

1 流處理技術(shù)概述

流處理系統(tǒng)能滿足對(duì)進(jìn)入系統(tǒng)的數(shù)據(jù)進(jìn)行即時(shí)計(jì)算的需要,相比Hadoop、Spark 等批處理系統(tǒng),在處理方式有非常大的不同.流處理更加像一個(gè)MapReduce計(jì)算的通用模型,只不過它的響應(yīng)時(shí)間可以達(dá)到秒級(jí)甚至是毫秒級(jí).流處理不需要對(duì)完整的數(shù)據(jù)樣本進(jìn)行計(jì)算,只針對(duì)通過系統(tǒng)的每一個(gè)數(shù)據(jù)項(xiàng)進(jìn)行操作.流處理系統(tǒng)理論上能夠處理無限多、無限大的數(shù)據(jù),但在同一個(gè)時(shí)間點(diǎn),卻只能夠處理一條(真正的流處理)或者少量(微批處理)的數(shù)據(jù),在不同記錄之間只保持最少量的狀態(tài).流處理屬于一般意義上的數(shù)據(jù)富集、持續(xù)處理,以及對(duì)于無界數(shù)據(jù)的分析過程的組合.流處理模式適合有近實(shí)時(shí)處理需求的任務(wù),如基于網(wǎng)站用戶行為實(shí)時(shí)產(chǎn)品推薦、經(jīng)營指標(biāo)實(shí)時(shí)計(jì)算、客戶信用審核、業(yè)務(wù)審核、反欺詐等.

目前,主要有兩種不同的方法來構(gòu)建流處理系統(tǒng),一種屬于真正的流處理(Native Streaming),所有被輸入的記錄或者事件都將按照它們進(jìn)入的先后順序被逐個(gè)處理,如Storm、Flink、Samza；另一種方式是微批處理(Micro-Batching),小的“批”由多條輸入的記錄組成,它們按照預(yù)設(shè)好的時(shí)間常量創(chuàng)建,通常是每隔幾秒生成一個(gè),如Spark Streaming、Trident-Storm.五種主流的流處理技術(shù)對(duì)比情況如表1所示.

其中,Storm、Spark Streaming 和Flink 是三種較常用的流處理技術(shù)[5].

表1 五種流處理技術(shù)對(duì)比情況

1.1 Storm[6]

Storm 比較適用于實(shí)時(shí)處理數(shù)據(jù)的場景,它是一套開源、分布式、高容錯(cuò)的實(shí)時(shí)計(jì)算系統(tǒng),在多個(gè)方面具備很強(qiáng)的優(yōu)勢.Storm 具有較強(qiáng)的容錯(cuò)性,可以對(duì)工作進(jìn)程及節(jié)點(diǎn)的故障進(jìn)行管理；流計(jì)算可以在多線程、進(jìn)程以及服務(wù)器之間并行展開,非常易于水平擴(kuò)展；Storm 的消息處理機(jī)制非常地可靠,不會(huì)遺漏信息,能保證每個(gè)消息都能得到完整的處理,而且在任務(wù)失敗時(shí)也能夠從消息源重試這個(gè)消息；底層使用MQ 作為消息隊(duì)列,能夠保證消息能得到快速的處理；Storm具有可靠的事務(wù)機(jī)制,即數(shù)據(jù)的處理完全精準(zhǔn),而且可以針對(duì)高峰、低峰時(shí)間段,動(dòng)態(tài)調(diào)整實(shí)時(shí)計(jì)算程序的并行度,以最大限度利用集群資源；同時(shí),Storm 的開發(fā)和單元測試也比較方便.Storm 目前發(fā)展的已經(jīng)相對(duì)比較成熟,部署和管理起來也很簡單,性能表現(xiàn)也十分出眾,常常被用于實(shí)時(shí)分析、持續(xù)計(jì)算、ETL、在線機(jī)器學(xué)習(xí)、分布式遠(yuǎn)程調(diào)用等.

1.2 Spark Streaming

Spark Streaming 能夠?qū)崿F(xiàn)對(duì)具有很高的吞吐量,需要高容錯(cuò)機(jī)制的流數(shù)據(jù)的實(shí)時(shí)處理,屬于Spark 核心API 的擴(kuò)展內(nèi)容之一,支持從Flume、Kafka、Twitter、Kinesis、ZeroMQ,以及TCP sockets 等多種數(shù)據(jù)源獲取數(shù)據(jù).獲取到數(shù)據(jù)源數(shù)據(jù)后,能夠利用map、reduce、join 和window 等高級(jí)函數(shù),處理特別復(fù)雜的算法,同時(shí)也可以把處理結(jié)果持久化到數(shù)據(jù)庫、文件系統(tǒng)、現(xiàn)場儀表盤等.但是與Storm 相比,Spark Streaming 適用于不要求實(shí)時(shí)處理和完全可靠的事務(wù)機(jī)制,不需要?jiǎng)討B(tài)調(diào)整并行度的場景.Spark Streaming 突出的優(yōu)點(diǎn)是吞吐量(即單位時(shí)間內(nèi)處理的數(shù)據(jù)量,MB/S)高,是Storm 的2-5 倍.如果除了要進(jìn)行實(shí)時(shí)計(jì)算外,還包含批處理(離線)、交互式查詢等需求,那么就要先選擇Spark 生態(tài),采用Spark Core 來實(shí)現(xiàn)批處理(離線)操作,使用Spark SQL 來實(shí)現(xiàn)交互式的查詢,再使用Spark Streaming 來實(shí)現(xiàn)流計(jì)算,將三者進(jìn)行無縫地整合,能夠給系統(tǒng)帶來非常高的擴(kuò)展性能.

1.3 Flink

Flink 目前還屬于新興的項(xiàng)目,仍處于不斷成熟的時(shí)期.它是介于Spark 和Storm 之間的一種架構(gòu),采用了原生的流處理系統(tǒng),與Spark Streaming 有相似的主從結(jié)構(gòu),與Storm 相似的數(shù)據(jù)流,所以Flink 兼具了低延遲和高吞吐的特性.同時(shí),Flink 在API 和容錯(cuò)性上也有很好的表現(xiàn),使用起來相對(duì)來說也比較簡單.Flink 具有許多特性：可進(jìn)行帶有事件時(shí)間的窗口(Window)操作；能進(jìn)行高吞吐、低延遲和高性能的流處理；窗口(Window)操作高度靈活；支持Exactlyonce(有狀態(tài)計(jì)算)語義；能進(jìn)行基于time、count、session,以及data-driven 的窗口操作；對(duì)于基于輕量級(jí)的分布式快照(Snapshot)實(shí)現(xiàn)可以容錯(cuò)；能運(yùn)行具備Backpressure 功能的持續(xù)流類模型；運(yùn)行時(shí)可同時(shí)支持Batch on Streaming 處理以及Streaming 處理；可進(jìn)行迭代計(jì)算；Flink 可在JVM 的內(nèi)部做到屬于自己的內(nèi)存管理；程序可自主優(yōu)化,這樣可以避免在非常特定情況下,產(chǎn)生Shuffle 及排序等高代價(jià)的操作,當(dāng)然中間結(jié)果需要進(jìn)行緩存操作.

2 基于Storm 的指標(biāo)實(shí)時(shí)計(jì)算方案

基于流處理技術(shù)的指標(biāo)實(shí)時(shí)計(jì)算,是通過實(shí)時(shí)監(jiān)聽和捕獲數(shù)據(jù)庫日志,利用流處理技術(shù)對(duì)日志和數(shù)據(jù)庫操作指令進(jìn)行實(shí)時(shí)解析,并實(shí)時(shí)將分析結(jié)果用于指標(biāo)計(jì)算的大數(shù)據(jù)處理模式.Flink 是新興的項(xiàng)目,而Storm 經(jīng)過多年發(fā)展已經(jīng)比較成熟,相較于Spark Streaming 的處理延時(shí)更低,甚至可以到毫秒級(jí),完全可以滿足指標(biāo)實(shí)時(shí)更新的需求,因此我們選擇Storm 作為實(shí)時(shí)處理的核心技術(shù).

Hadoop 是一個(gè)批處理系統(tǒng),它由于具備數(shù)據(jù)吞吐量大、自動(dòng)容錯(cuò)等很多優(yōu)點(diǎn),所以廣泛應(yīng)用在海量異構(gòu)大數(shù)據(jù)的處理上.但是,Hadoop 適合大批量數(shù)據(jù)的離線處理,并不擅長實(shí)時(shí)計(jì)算,因?yàn)樗緛砭褪菫榕幚矶_發(fā)的,這也是大家一致的共識(shí).不過Hadoop 卻可以作為Storm 等組件運(yùn)行的基礎(chǔ)框架平臺(tái).因此,整個(gè)實(shí)時(shí)計(jì)算系統(tǒng)基于Hadoop 平臺(tái)構(gòu)建,由日志采集、消息管理(Kafka)、協(xié)調(diào)管理(Zookeeper)、實(shí)時(shí)處理(Storm)、內(nèi)存數(shù)據(jù)庫(Redis)等部分構(gòu)成[7].其中,由日志采集模塊(使用Shell、C、Java 等腳本開發(fā)的插件)監(jiān)聽數(shù)據(jù)庫日志,并實(shí)時(shí)把日志抓取下來推送至Kafka 分布式消息管理系統(tǒng)；通過Storm 系統(tǒng)消費(fèi)Kafka 中的消息,同時(shí)通過Zookeeper 管理期間的消費(fèi)記錄；由Storm 根據(jù)指標(biāo)計(jì)算邏輯對(duì)日志進(jìn)行定制化分析和處理,并輸出到Redis 內(nèi)存數(shù)據(jù)庫中；最后由應(yīng)用程序讀取Redis 中的結(jié)果并展示給用戶,或轉(zhuǎn)入數(shù)據(jù)庫進(jìn)行持久化存儲(chǔ).從技術(shù)架構(gòu)看,自上而下,首先是源端數(shù)據(jù)庫,下一層是日志采集部分,可以針對(duì)多個(gè)數(shù)據(jù)庫同時(shí)進(jìn)行采集,日志采集之下是Kafka 消息管理系統(tǒng),Kafka 消息管理系統(tǒng)層之下是Storm 實(shí)時(shí)處理層,Kafka 和Storm 之間的協(xié)調(diào)管理由Zookeeper 承擔(dān),Storm 流處理層之下是Redis 內(nèi)存數(shù)據(jù)庫,最下層是Web 或者App 應(yīng)用,也可以包括用于持久化的數(shù)據(jù)庫(如Hbase 等列式數(shù)據(jù)庫),詳情如圖1所示.

2.1 日志采集

日志采集程序或工具有多種可選方式.在此列舉三種,一種是編寫Shell、C 或Java 腳本程序,自編腳本輕量和完全自主可控,對(duì)服務(wù)器產(chǎn)生的壓力相對(duì)較小,但需要一定的自主開發(fā)量；第二種是采用第三方框架技術(shù)直接進(jìn)行采集,比如采用Flume.Flume 屬于分布式的、高效的日志采集系統(tǒng),可以把分布在不同服務(wù)器上的海量日志文件統(tǒng)一收集到一個(gè)集中的存儲(chǔ)資源中,但是Flume 的配置卻不怎么簡單,Source、Channel、Sink 的關(guān)系交織在配置文件中的話,非常不便于管理；還有一種方式是使用CDC(Change Data Capture)產(chǎn)品實(shí)現(xiàn)源端數(shù)據(jù)庫日志的采集,但CDC 通常需要進(jìn)行單獨(dú)采購,同時(shí)需要在源端數(shù)據(jù)庫和目標(biāo)端數(shù)據(jù)安裝軟件,成本較高,對(duì)服務(wù)器性能要求也高.

圖1 基于Storm 的實(shí)時(shí)分析系統(tǒng)技術(shù)架構(gòu)

本方案采用自行開發(fā)的腳本程序進(jìn)行采集,直接將采集腳本部署在源端數(shù)據(jù)庫服務(wù)器,實(shí)時(shí)監(jiān)聽和讀取磁盤設(shè)備中的日志文件.此方式相較于另外兩種方式有三個(gè)優(yōu)勢：一是部署簡單,直接部署在源端數(shù)據(jù)庫服務(wù)器即可,無需額外地搭建同步服務(wù)器,Flume 和CDC 都需要若干臺(tái)服務(wù)器來部署；二是響應(yīng)速度快,可實(shí)現(xiàn)實(shí)時(shí)捕獲增量日志,而CDC 的實(shí)時(shí)同步間隔通常是數(shù)秒,無法滿足要求；三是只涉及磁盤文件讀取,占用源端服務(wù)器的CPU 和內(nèi)存資源少,對(duì)服務(wù)器的運(yùn)行影響不大.

2.2 消息管理(Kafka)

Kafka 是基于日志文件的消息系統(tǒng),消息能夠持久化存儲(chǔ)到硬盤中,數(shù)據(jù)不容易丟失.Kafka 可以保存消息的進(jìn)度及位置,對(duì)于用戶來說,也可以自行定義消費(fèi)的起始點(diǎn),可以實(shí)現(xiàn)消息的重復(fù)和多次消費(fèi).Kafka 同時(shí)具有隊(duì)列和發(fā)布訂閱兩種消息消費(fèi)模式,可以保證消息隊(duì)列中的消息能按照順序被消費(fèi)并且與Storm 的契合度很高.此外,Kafka 的Consumer 是pull-based 模型,該模型可以緩解日志產(chǎn)生速度快于消費(fèi)速度的壓力,使消費(fèi)速度合理匹配生產(chǎn)速度.把Kafka 消息系統(tǒng)放置在日志采集和Storm 模塊中間,是防止在突發(fā)的、高并發(fā)的情況之下,由于日志可能會(huì)出現(xiàn)井噴式的增長,如果這時(shí)候Storm 的消費(fèi)速度不能快于日志的產(chǎn)生速度,就會(huì)導(dǎo)致大量消息處理滯后,進(jìn)而導(dǎo)致丟失,所以加入了Kafka 消息系統(tǒng)作為數(shù)據(jù)緩沖區(qū).

2.3 協(xié)調(diào)管理(Zookeeper)

Zookeeper 是一個(gè)針對(duì)分布式系統(tǒng)的高可靠地協(xié)調(diào)系統(tǒng),它可以讓分布式系統(tǒng)在大多數(shù)情況下正常運(yùn)行.一是可以提供分布式的鎖服務(wù)[8].分布式集群系統(tǒng)中,讀取與分析等操作會(huì)分散到不同的節(jié)點(diǎn)之上進(jìn)行,所以在數(shù)據(jù)操作的過程中就有可能發(fā)生一致性問題.Zookeeper 提供的這種鎖服務(wù)就很好地解決了此問題,保證了進(jìn)行分布式數(shù)據(jù)運(yùn)算時(shí)的數(shù)據(jù)操作的一致性；二是能夠?yàn)榉植际降南到y(tǒng)提供故障恢復(fù)的支持.Storm中master 節(jié)點(diǎn)運(yùn)行的守護(hù)進(jìn)程“Nimbus”和worker 節(jié)點(diǎn)運(yùn)行的守護(hù)進(jìn)程“Supervisor”之間的協(xié)調(diào)工作是通過Zookeeper 來管理的,Nimbus 和Supervisor 自身在集群上都是無狀態(tài)的,它們的狀態(tài)都保存在Zookeeper中,所以任何節(jié)點(diǎn)的宕機(jī)和動(dòng)態(tài)擴(kuò)容都不會(huì)影響整個(gè)集群的工作運(yùn)行；三是Zookeeper 也可以管理Kafka 的消費(fèi)記錄,即使遭遇Kafka 宕機(jī),在進(jìn)行重啟之后也能定位上次的消費(fèi)記錄,從宕機(jī)點(diǎn)繼續(xù)進(jìn)行消費(fèi),實(shí)現(xiàn)了“斷點(diǎn)續(xù)傳”.

2.4 實(shí)時(shí)處理(Storm)

Storm 能夠相對(duì)比較簡單地實(shí)現(xiàn)對(duì)復(fù)雜實(shí)時(shí)計(jì)算的編寫以及擴(kuò)展.數(shù)據(jù)庫數(shù)據(jù)的實(shí)時(shí)處理會(huì)使用Storm,就像好比離線數(shù)據(jù)批處理常常使用Hadoop 一樣,而且Storm 能保證沒有遺漏,保證每一個(gè)消息都能被處理,速度也比較快.在一個(gè)相對(duì)較小的集群中,可以使用多種語言編程,如使用Java、Payson 等語言進(jìn)行開發(fā),每秒能處理百萬級(jí)別的消息.Storm 作為整個(gè)指標(biāo)實(shí)時(shí)計(jì)算模式的功能核心和技術(shù)核心部分,主要完成三個(gè)方面的工作,即日志解析、指令解析、實(shí)時(shí)計(jì)算.該部分的日志處理能力主要受單一數(shù)據(jù)庫只能采用單線程處理的限制,實(shí)戰(zhàn)時(shí)要注意避免該問題成為整體處理能力的提升的瓶頸,但是不同的數(shù)據(jù)庫日志可以并發(fā)處理.

2.4.1 日志解析不同類型的數(shù)據(jù)庫產(chǎn)品的日志編碼規(guī)則和存儲(chǔ)邏輯各不相同,解析日志需要首先研究數(shù)據(jù)庫日志的編碼和存儲(chǔ)等規(guī)則,這是整個(gè)計(jì)算模式能否正常運(yùn)行的前提之一,否則無法將日志解析和轉(zhuǎn)換為易于識(shí)別的信息.日志解析程序在接收到日志消息后,將根據(jù)數(shù)據(jù)庫的日志規(guī)則,自動(dòng)切分日志,識(shí)別日志類型,剔除回滾等不改變數(shù)據(jù)的日志類型,僅保留增、刪、改等操作產(chǎn)生的日志,并將該部分日志的每一頁內(nèi)容由十六進(jìn)制編碼轉(zhuǎn)換為“標(biāo)準(zhǔn)和可用”數(shù)據(jù)用于下一步進(jìn)行指令解析.整個(gè)日志解析過程可劃分為捕獲、切分、識(shí)別和轉(zhuǎn)換等四個(gè)部分,如圖2所示.

圖2 日志解析過程圖

2.4.2 指令解析

基于日志解析部分的結(jié)果,指令解析部分將按照擬統(tǒng)計(jì)的大數(shù)據(jù)指標(biāo)的算法要求,從日志中篩選出指標(biāo)計(jì)算涉及到的所有操作信息,解析每個(gè)增、刪、改等操作涉及的指令所影響的數(shù)據(jù)表以及字段信息,抽取出用于數(shù)據(jù)篩選和計(jì)算的信息,尤其是相應(yīng)的數(shù)據(jù)增量變化信息.簡單來說,就是通過指令解析從中獲取指標(biāo)計(jì)算所需要的全部信息,然后將解析結(jié)果推送至內(nèi)存計(jì)算程序進(jìn)行下一步處理.指令解析過程可劃分為識(shí)別、篩選、解析和推送等四個(gè)部分,如圖3所示.

圖3 指令解析過程圖

2.4.3 實(shí)時(shí)計(jì)算

以數(shù)據(jù)統(tǒng)計(jì)類指標(biāo)計(jì)算為例.傳統(tǒng)的處理方式是首先同步源端數(shù)據(jù)庫和目標(biāo)端數(shù)據(jù)庫的數(shù)據(jù),待增量數(shù)據(jù)在目標(biāo)端數(shù)據(jù)庫同步且入庫完成之后,再調(diào)用程序根據(jù)指標(biāo)算法對(duì)全量樣本數(shù)據(jù)進(jìn)行聚合計(jì)算,計(jì)算結(jié)果保存到數(shù)據(jù)庫中等待應(yīng)用讀取,數(shù)據(jù)的時(shí)效性、連續(xù)性比較差.基于流處理的實(shí)時(shí)計(jì)算則是對(duì)數(shù)據(jù)庫數(shù)據(jù)進(jìn)行實(shí)時(shí)、在線、同步處理,無需入庫,無需對(duì)樣本進(jìn)行全量計(jì)算,直接在上一次計(jì)算結(jié)果基礎(chǔ)上進(jìn)行處理,理論上可提供毫秒級(jí)的實(shí)時(shí)計(jì)算能力.

實(shí)時(shí)計(jì)算程序需要根據(jù)指標(biāo)算法預(yù)設(shè)某指標(biāo)的完整計(jì)算規(guī)則,基于日志和指令解析的結(jié)果,自動(dòng)適配該指標(biāo)的計(jì)算邏輯.在篩選出需要參與計(jì)算的信息后,針對(duì)Insert、Update、Delete 指令選擇相應(yīng)的處理方式進(jìn)行計(jì)算.如果是Insert 插入指令,可以在上一次統(tǒng)計(jì)結(jié)果基礎(chǔ)上直接進(jìn)行“加法”操作；Update 更新指令在日志中會(huì)記錄該表原始數(shù)據(jù)塊和該表最新數(shù)據(jù)塊,可根據(jù)數(shù)據(jù)實(shí)際變化情況進(jìn)行“加法或者減法”操作；Delete 刪除指令進(jìn)行“減法”操作.對(duì)上一次計(jì)算得出的結(jié)果進(jìn)行相應(yīng)的“加法或者減法”計(jì)算后,將得出的最新數(shù)值寫入內(nèi)存數(shù)據(jù)庫中供應(yīng)用調(diào)用.當(dāng)然,也可以保存所有增量變化的信息,持久化到備份庫或者寬表中滿足不同場景的需要.

有些指標(biāo)的計(jì)算需要考慮復(fù)雜的篩選條件,有的可能需要進(jìn)行復(fù)合或混合運(yùn)算,只要獲取了“字段”的變化情況,無非是計(jì)算的復(fù)雜度得到了增加,但是復(fù)雜的判斷和計(jì)算勢必會(huì)影響實(shí)時(shí)處理的效率,這點(diǎn)需要在數(shù)據(jù)庫建模時(shí)統(tǒng)籌進(jìn)行考慮,并在處理邏輯上進(jìn)行優(yōu)化.此外,需要注意的是,內(nèi)存數(shù)據(jù)庫中的數(shù)據(jù)應(yīng)當(dāng)定期進(jìn)行轉(zhuǎn)儲(chǔ),比如可以將轉(zhuǎn)儲(chǔ)的數(shù)據(jù)保存至HBase 列存儲(chǔ)數(shù)據(jù)庫中,這樣就可以在系統(tǒng)宕機(jī)后對(duì)數(shù)據(jù)重新進(jìn)行初始化.

2.5 方案優(yōu)勢

上述基于流處理技術(shù)的指標(biāo)實(shí)時(shí)計(jì)算方案,采用了較成熟的主流技術(shù)和工具,能夠?qū)崿F(xiàn)對(duì)大數(shù)據(jù)的實(shí)時(shí)、在線以及持續(xù)地處理,可以滿足業(yè)務(wù)和管理決策對(duì)數(shù)據(jù)的實(shí)時(shí)性需求.該方案與傳統(tǒng)數(shù)據(jù)倉庫、BI、數(shù)據(jù)采集(如CDC 等)等技術(shù)相比,具有五大優(yōu)勢.一是處理高效,從捕獲數(shù)據(jù)庫日志到完成指標(biāo)實(shí)時(shí)計(jì)算的時(shí)耗能達(dá)到毫秒級(jí)；二是對(duì)源端數(shù)據(jù)庫服務(wù)器影響小,因直接讀取服務(wù)器磁盤日志文件,不涉及數(shù)據(jù)庫系統(tǒng)級(jí)的管理和交互,所以基本不占用源端服務(wù)器的CPU 和內(nèi)存資源；三是可靠性高,消息隊(duì)列(kafka)和協(xié)調(diào)系統(tǒng)(Zookeeper)保障了日志能夠逐條被處理,并且整個(gè)集群在宕機(jī)后能夠快速恢復(fù)；四是成本低,流處理集群基于X86 架構(gòu)服務(wù)器搭建,價(jià)格低,采購、維護(hù)和升級(jí)簡單；五是可擴(kuò)展性強(qiáng),整個(gè)系統(tǒng)采用Hadoop分布式集群架構(gòu),通過增加硬件設(shè)備可實(shí)現(xiàn)處理能力的線性提升.

3 應(yīng)用實(shí)踐

3.1 流處理技術(shù)實(shí)戰(zhàn)

基于Storm 的大數(shù)據(jù)指標(biāo)實(shí)時(shí)計(jì)算模式已經(jīng)在某省級(jí)單位進(jìn)行了實(shí)踐.數(shù)據(jù)庫產(chǎn)品的型號(hào)及版本為IBM Informix 11,在局域網(wǎng)(千兆)內(nèi)架設(shè)基于X86 架構(gòu)PC 服務(wù)器的Hadoop(開源)集群作為運(yùn)行平臺(tái).日志采集使用Java 插件完成,采用Hadoop、Kafka、Zookeeper、Storm、Redis 等開源產(chǎn)品和組件完成消息管理、協(xié)調(diào)管理、流計(jì)算、內(nèi)存計(jì)算等工作.整個(gè)實(shí)戰(zhàn)環(huán)境不超過10 臺(tái)PC 服務(wù)器,可配置4-6 個(gè)計(jì)算節(jié)點(diǎn)和2-4 個(gè)管理節(jié)點(diǎn),主要用于處理本地的兩個(gè)Informix 數(shù)據(jù)庫日志.基于以上的環(huán)境和配置構(gòu)建的流處理計(jì)算系統(tǒng),實(shí)現(xiàn)了兩個(gè)大類,不少于3 個(gè)維度的統(tǒng)計(jì)指標(biāo)的實(shí)時(shí)計(jì)算,達(dá)到了毫秒級(jí)的準(zhǔn)實(shí)時(shí)計(jì)算效果.值得一提的是,在每個(gè)數(shù)據(jù)庫日志200-500 MB 大小的情況下,各臺(tái)服務(wù)器CPU 占用率僅為5%左右,剩余空閑資源可利用空間還非常大,還可以增加更多的指標(biāo)進(jìn)行處理,當(dāng)然也要考慮日志解析節(jié)點(diǎn)的綜合處理能力,不能盲目的增加過多的計(jì)算內(nèi)容.

3.2 應(yīng)用成果

該單位的數(shù)據(jù)發(fā)布平臺(tái)對(duì)接了流處理系統(tǒng)的實(shí)時(shí)指標(biāo)數(shù)據(jù).依托于流處理分布式集群的快速處理能力,流處理系統(tǒng)的計(jì)算結(jié)果通過消息機(jī)制實(shí)時(shí)推送至發(fā)布平臺(tái).以該單位的保險(xiǎn)費(fèi)指標(biāo)為例,實(shí)時(shí)數(shù)據(jù)支持日期、機(jī)構(gòu)和產(chǎn)品三個(gè)維度,可滿足各級(jí)管理人員實(shí)時(shí)監(jiān)控和分析業(yè)務(wù)情況的需要.下一步,該單位計(jì)劃逐步開發(fā)賠款、賠案、實(shí)收保費(fèi)、應(yīng)收保費(fèi)等更多指標(biāo)的實(shí)時(shí)展示功能,進(jìn)一步增強(qiáng)平臺(tái)的業(yè)務(wù)和管理支撐能力.

4 存在的主要問題

目前,基于Storm 的大數(shù)據(jù)指標(biāo)實(shí)時(shí)計(jì)算方法存在兩個(gè)相對(duì)比較大的問題.一是數(shù)據(jù)庫產(chǎn)品升級(jí)可能帶來日志格式的變化.數(shù)據(jù)庫產(chǎn)品手冊并未說明日志的編碼規(guī)則,日志的類型等信息需要自行研究,如果數(shù)據(jù)庫版本變化導(dǎo)致日志編碼或存儲(chǔ)規(guī)則發(fā)生變化,這種情況就需要在升級(jí)之前重新研究日志規(guī)則,然后對(duì)應(yīng)調(diào)整日志解析算法；二是指標(biāo)口徑調(diào)整可能引起系統(tǒng)處理邏輯的變更.如果指標(biāo)口徑調(diào)整導(dǎo)致算法邏輯發(fā)生變化,比如統(tǒng)計(jì)的字段或者數(shù)據(jù)篩選條件發(fā)生改變,就需要調(diào)整指令解析和實(shí)時(shí)計(jì)算程序.這兩種情況尤其是第二種問題如果頻繁發(fā)生,可能要耗費(fèi)大量的時(shí)間和人力成本完成相應(yīng)改造工作.

5 結(jié)束語

基于Storm 的大數(shù)據(jù)指標(biāo)實(shí)時(shí)計(jì)算方法,尚處于研究階段,仍需要進(jìn)一步的測試和優(yōu)化,穩(wěn)定性有待進(jìn)一步提高,處理能力也還有挖掘的空間.就現(xiàn)階段應(yīng)用實(shí)踐效果來看,在不需要大量資金投入的情況下,滿足數(shù)據(jù)規(guī)模適中的企業(yè)的少量指標(biāo)的實(shí)時(shí)計(jì)算基本沒有問題,但是數(shù)據(jù)日志規(guī)則研究和單個(gè)數(shù)據(jù)庫日志的解析效率問題,目前仍然是實(shí)現(xiàn)大批量指標(biāo)計(jì)算的掣肘,所以大規(guī)模應(yīng)用的基礎(chǔ)仍需要進(jìn)一步的夯實(shí).但是隨著大數(shù)據(jù)技術(shù)日新月異的不斷進(jìn)步,更成熟和強(qiáng)大的組件或產(chǎn)品也會(huì)不斷涌現(xiàn),該技術(shù)在將來通過持續(xù)地升級(jí)和調(diào)整,流處理能力也必將會(huì)越來越強(qiáng)大.