引言

在大數(shù)據(jù)的生態(tài)系統(tǒng)里，時(shí)間序列數(shù)據(jù)(Time Series Data，簡(jiǎn)稱(chēng)TSD)是很常見(jiàn)也是所占比例最大的一類(lèi)數(shù)據(jù)，幾乎出現(xiàn)在科學(xué)和工程的各個(gè)領(lǐng)域，一些常見(jiàn)的時(shí)間序列數(shù)據(jù)有：描述服務(wù)器運(yùn)行狀況的Metrics數(shù)據(jù)、各種IoT系統(tǒng)的終端數(shù)據(jù)、腦電圖、匯率、股價(jià)、氣象和天文數(shù)據(jù)等等，時(shí)序數(shù)據(jù)在數(shù)據(jù)特征和處理方式上有很大的共性，因此也催生了一些面向面向時(shí)序數(shù)據(jù)的特定工具，比如時(shí)序數(shù)據(jù)庫(kù)和時(shí)序數(shù)據(jù)可視化工具等等，在云平臺(tái)上也開(kāi)始出現(xiàn)面向時(shí)序數(shù)據(jù)的SaaS/PaaS服務(wù)，例如微軟最近剛剛發(fā)布的Azure Time Series Insight。本文會(huì)介紹一個(gè)時(shí)間序列數(shù)據(jù)處理平臺(tái)案例，探討這類(lèi)大數(shù)據(jù)平臺(tái)在架構(gòu)、選型和設(shè)計(jì)上的一些實(shí)踐經(jīng)驗(yàn)以供參考。

業(yè)務(wù)場(chǎng)景

本文介紹的案例是一個(gè)面向大型企業(yè)IT系統(tǒng)運(yùn)維的監(jiān)控平臺(tái)，數(shù)據(jù)來(lái)源于多種監(jiān)控終端產(chǎn)生的時(shí)序數(shù)據(jù)，涉及的數(shù)據(jù)源涵蓋了SCOM、AppDynamics、Website Pulse、Piwik以及AWS Cloud Watch等多種主流的第三方監(jiān)控工具，基于組織內(nèi)部的IT規(guī)范，所有應(yīng)用系統(tǒng)都安裝了上述一種或多種監(jiān)控工具，這為建立一個(gè)統(tǒng)一的多維度的監(jiān)控平臺(tái)提供了保證，該平臺(tái)基于多種監(jiān)控?cái)?shù)據(jù)，對(duì)同一應(yīng)用/服務(wù)系統(tǒng)進(jìn)行綜合的健康評(píng)估，在發(fā)生故障時(shí)會(huì)根據(jù)不同的數(shù)據(jù)源進(jìn)行交叉驗(yàn)證，從而幫助運(yùn)維人員快速和準(zhǔn)確地定位故障原因。

架構(gòu)設(shè)計(jì)

完整的大數(shù)據(jù)系統(tǒng)往往包涵數(shù)據(jù)采集，消息對(duì)列，實(shí)時(shí)流處理，離線批處理，數(shù)據(jù)存儲(chǔ)和數(shù)據(jù)展示等多個(gè)組件，為了滿足業(yè)務(wù)上對(duì)實(shí)時(shí)監(jiān)控和歷史數(shù)據(jù)匯總分析的需求，系統(tǒng)遵循了Lambda架構(gòu)，將實(shí)時(shí)流處理與離線批處理進(jìn)行了分離。此外，鑒于平臺(tái)處理的所有數(shù)據(jù)均為時(shí)序數(shù)據(jù)，在架構(gòu)上針對(duì)這個(gè)特點(diǎn)著重進(jìn)行了調(diào)整和優(yōu)化，其中重要的一環(huán)是引入“時(shí)間序列數(shù)據(jù)庫(kù)”作為核心的數(shù)據(jù)存儲(chǔ)與查詢(xún)引擎。

系統(tǒng)完整的數(shù)據(jù)流如下：首先，數(shù)據(jù)被數(shù)據(jù)采集組件從外部系統(tǒng)采集并來(lái)放入消息隊(duì)列，接著，流處理組件從隊(duì)列中取出數(shù)據(jù)進(jìn)行流式計(jì)算，消息隊(duì)列從中的起到的作用是平衡“生產(chǎn)者”——數(shù)據(jù)采集組件和“消費(fèi)者”——流處理組件在消息處理上的速率差，提升系統(tǒng)的穩(wěn)定性和可靠性。數(shù)據(jù)在流處理組件中會(huì)經(jīng)歷清洗、過(guò)濾、轉(zhuǎn)換、業(yè)務(wù)處理等諸多環(huán)節(jié)，之后按TSD引擎規(guī)定的標(biāo)準(zhǔn)TSD格式推送到TSD引擎，由TSD引擎最終寫(xiě)入后端數(shù)據(jù)庫(kù)。

實(shí)時(shí)流處理部分要求數(shù)據(jù)從采集到最后的展示控制在秒級(jí)延遲，嚴(yán)格來(lái)說(shuō)，這是一套近實(shí)時(shí)系統(tǒng)，但其實(shí)時(shí)性已經(jīng)足夠滿足業(yè)務(wù)上的需求，為了保證處理速率，實(shí)時(shí)鏈條上的數(shù)據(jù)大多數(shù)時(shí)間是駐留在內(nèi)存中的，好在實(shí)時(shí)部分只關(guān)注近兩周的數(shù)據(jù)，所以總的內(nèi)存消耗處在可控的范圍之內(nèi)。

在批處理數(shù)據(jù)線上，利用數(shù)據(jù)庫(kù)的同步機(jī)制將實(shí)時(shí)部分落地的數(shù)據(jù)持續(xù)同步到批處理的數(shù)據(jù)庫(kù)上，這個(gè)庫(kù)存儲(chǔ)著數(shù)據(jù)全集，所有批處理相關(guān)的查詢(xún)都在這個(gè)庫(kù)上執(zhí)行，與實(shí)時(shí)部分的組件完全隔離。批處理會(huì)保存過(guò)去三年的數(shù)據(jù)，分析尺度多為日，周，月甚至年。不同于一般離線分析系統(tǒng)選型Hive一類(lèi)的數(shù)據(jù)倉(cāng)庫(kù)，我們希望在離線分析時(shí)繼續(xù)充分利用時(shí)序數(shù)據(jù)庫(kù)帶來(lái)的種種好處，比如經(jīng)過(guò)特殊優(yōu)化的時(shí)序數(shù)據(jù)查詢(xún)，開(kāi)箱即用的查詢(xún)接口等等，所以在離線部分我們依然配備TSD引擎，批處理組件在實(shí)現(xiàn)業(yè)務(wù)需求時(shí)可以深度利用TSD引擎對(duì)時(shí)序數(shù)據(jù)進(jìn)行聚合運(yùn)算，在聚合之后的結(jié)果上再進(jìn)行更加復(fù)雜的分析并寫(xiě)回?cái)?shù)據(jù)庫(kù)，同時(shí)也可以在普通查詢(xún)無(wú)法實(shí)現(xiàn)需求時(shí)越過(guò)TSD引擎直接對(duì)底層數(shù)據(jù)文件進(jìn)行MR計(jì)算。

最后，數(shù)據(jù)展示組件會(huì)從TSD引擎中提取數(shù)據(jù)，以各種形式的圖表展示給用戶。在實(shí)際的開(kāi)發(fā)中我們發(fā)現(xiàn)TSD引擎對(duì)數(shù)據(jù)格式有諸多的限制，有的TSD需要進(jìn)行某些轉(zhuǎn)換和適配才能展示，因此我們?cè)赥SD引擎和數(shù)據(jù)展示組件中間引入了一個(gè)輕量的驅(qū)動(dòng)程序來(lái)透明地解決這些問(wèn)題。

基于上述分析和實(shí)際的原型驗(yàn)證，在多輪迭代之后，我們最終成形的系統(tǒng)架構(gòu)如下：

接下去我們會(huì)對(duì)每個(gè)組件逐一進(jìn)行介紹。

組件與選型

數(shù)據(jù)采集

平臺(tái)的數(shù)據(jù)來(lái)源非常多，涉及到的協(xié)議類(lèi)型自然就多，并且伴隨著以后的持續(xù)建設(shè)，會(huì)有越來(lái)越多新的數(shù)據(jù)源和傳輸協(xié)議需要被支持，因此我們希望選定的組件能夠支持非常豐富的協(xié)議類(lèi)型，同時(shí)盡可能地通過(guò)配置去集成數(shù)據(jù)源并采集數(shù)據(jù)，避免編寫(xiě)大量的代碼。目前業(yè)界較為主流的數(shù)據(jù)采集工具有Flume、Logstash以及Kafka Connect等等，這些工具各有各的特點(diǎn)和擅長(zhǎng)領(lǐng)域，但是在支持協(xié)議的豐富性和可配置性上，與我們的需求有一定的差距。

其實(shí)有一個(gè)一直被人忽視但卻是非常理想的數(shù)據(jù)采集組件——Apache Camel，它主要應(yīng)用于企業(yè)應(yīng)用集成領(lǐng)域，也被一些系統(tǒng)作為ESB(企業(yè)服務(wù)總線)使用，其作用是在應(yīng)用系統(tǒng)林立的企業(yè)IT環(huán)境中扮演一個(gè)“萬(wàn)向接頭”的角色，讓數(shù)據(jù)和信息在各種不同的系統(tǒng)間平滑地交換和流轉(zhuǎn)，經(jīng)過(guò)多年的積累，Camel已經(jīng)支持近200種協(xié)議或數(shù)據(jù)源，并且可以完全基于配置實(shí)現(xiàn)，這恰好滿足了我們數(shù)據(jù)采集的需求，經(jīng)過(guò)原型驗(yàn)證，也證明了我們的選擇是明智的。

最后，作為一個(gè)非大數(shù)據(jù)組件，對(duì)于Camel的性能和吞吐量我們是有清晰認(rèn)識(shí)的，通過(guò)對(duì)數(shù)據(jù)源進(jìn)行分組，使用多個(gè)Camel實(shí)例分區(qū)采集數(shù)據(jù)，我們從架構(gòu)上輕松地解決了這些問(wèn)題。

消息隊(duì)列

在消息隊(duì)列的選擇上沒(méi)有可以討論的，Kafka幾乎是不二的選擇，我們也不例外。

流處理

流處理和批處理都是業(yè)務(wù)邏輯最集中的地方，也是系統(tǒng)的核心。目前用于流處理的主流技術(shù)是Storm和Spark Streaming，對(duì)兩者進(jìn)行比較的文章很多，通常認(rèn)為Storm具有更高的實(shí)時(shí)性，可以做到最低亞秒級(jí)的延遲，相比之下Spark Streaming的實(shí)時(shí)性要差一些，因?yàn)樗浴眒icro batch”的方式進(jìn)行流處理的，但是依托Spark這個(gè)大平臺(tái)，從統(tǒng)一技術(shù)堆棧和與其他Spark組件交互的角度考慮，Spark Streaming變得越來(lái)越流行，鑒于在業(yè)務(wù)上秒級(jí)延遲已經(jīng)可以滿足需求，我們最終選擇了后者。

批處理

傳統(tǒng)大數(shù)據(jù)的離線處理多選擇以Hive為代表的數(shù)據(jù)倉(cāng)庫(kù)進(jìn)行建模和分析，這在很多項(xiàng)目上被證明是可靠的解決方案。后來(lái)隨著Spark的不斷壯大，Spark SQL的使用越來(lái)越廣泛，并且Spark SQL完全兼容Hive，這使得遷移工作幾乎沒(méi)有任何障礙。對(duì)于復(fù)雜的非結(jié)構(gòu)化數(shù)據(jù)，Hadoop平臺(tái)上通過(guò)MR編程去處理，Spark是通過(guò)Spark Core的RDD編程實(shí)現(xiàn)。如今Spark在大數(shù)據(jù)處理的很多方面已經(jīng)取代Hadoop成為大數(shù)據(jù)的首選技術(shù)平臺(tái)，我們?cè)谂幚淼倪x型上也沒(méi)有過(guò)多的討論，使用Spark Core + Spark SQL是一個(gè)自然而然的決定。

但是考慮到系統(tǒng)處理的是TSD數(shù)據(jù)，如前文所屬，在批處理的數(shù)據(jù)鏈條上，TSD引擎依然是一個(gè)必不可少的角色，我們?cè)O(shè)計(jì)的策略是:

所有TSD引擎可以直接支持的查詢(xún)交由TSD引擎直接處理

復(fù)雜的業(yè)務(wù)處理可以通過(guò)TSD引擎進(jìn)行預(yù)處理，將預(yù)處理結(jié)果交給Spark Core進(jìn)行深度分析并將結(jié)果寫(xiě)回?cái)?shù)據(jù)庫(kù)

針對(duì)TSD引擎無(wú)法完成的分析邏輯，由Spark Core或Spark SQL繞過(guò)TSD引擎，直連后端的HBase進(jìn)行分析處理，結(jié)果同樣直接寫(xiě)到HBase上

為提升性能，對(duì)分析中使用到的以日/周/月/年為單位的中間表進(jìn)行預(yù)生成計(jì)算。

主數(shù)據(jù)管理

主數(shù)據(jù)是指來(lái)自數(shù)據(jù)源的核心業(yè)務(wù)對(duì)象，對(duì)于我們這個(gè)以監(jiān)控為核心的平臺(tái)，主數(shù)據(jù)包括：服務(wù)器、系統(tǒng)拓?fù)浣Y(jié)構(gòu)、站點(diǎn)、網(wǎng)絡(luò)設(shè)施等等，主數(shù)據(jù)往往都跨越多種不同的數(shù)據(jù)源，并且經(jīng)常發(fā)生變更，需要對(duì)其進(jìn)行定期維護(hù)。

為此，我們構(gòu)建了一個(gè)統(tǒng)一的主數(shù)據(jù)管理組件，并通過(guò)Web Service的方式向外提供主數(shù)據(jù)，由于平臺(tái)在流處理和批處理過(guò)程中需要頻繁地使用主數(shù)據(jù)，而主數(shù)據(jù)的體量并不大，所以我們會(huì)讓流處理和批處理組件一次性地將主數(shù)據(jù)加載到內(nèi)存中，同時(shí)為它們加入命令行和Restful API接口，允許它們?cè)谥鲾?shù)據(jù)發(fā)生變更時(shí)重新加載主數(shù)據(jù)。

主數(shù)據(jù)管理模塊是一個(gè)傳統(tǒng)的Web應(yīng)用，基于Spring-Boot構(gòu)建，使用MySQL存儲(chǔ)導(dǎo)入的主數(shù)據(jù)，對(duì)外通過(guò)Restful API提供主數(shù)據(jù)供給服務(wù)，它還有一個(gè)管理頁(yè)面方便管理員維護(hù)主數(shù)據(jù)。

TSD引擎與數(shù)據(jù)存儲(chǔ)

TSD引擎負(fù)責(zé)TSD的寫(xiě)入和查詢(xún)，很多TSD數(shù)據(jù)庫(kù)會(huì)利用一個(gè)成熟的NoSQL數(shù)據(jù)庫(kù)進(jìn)行數(shù)據(jù)存儲(chǔ)，而TSD引擎則專(zhuān)注在TSD數(shù)據(jù)的處理上。這兩部分密不可分，因此我們放在一起討論。

我們對(duì)時(shí)間序列數(shù)據(jù)庫(kù)的選型主要是在目前業(yè)界最主流的兩個(gè)產(chǎn)品InfluxDB和OpenTSDB之間展開(kāi)的。 前者使用GO語(yǔ)言編寫(xiě)，后端存儲(chǔ)先后使用過(guò)LevelDB和BoltDB，現(xiàn)在使用的則是InfluxDB自己實(shí)現(xiàn)的Time Structured Merge Tree引擎，OpenTSDB使用Java編寫(xiě)，后端存儲(chǔ)使用HBase。在單機(jī)性能上，多種對(duì)比測(cè)試顯示InfluxDB具有更高的性能，但我們最終選擇的是OpenTSDB，主要原因是考慮到在集群和水平伸縮方面，背靠HBase的OpenTSDB有明顯的優(yōu)勢(shì)，相比之下InfluxDB只在收費(fèi)的企業(yè)版提供集群功能，同時(shí)在集群規(guī)模和支撐的數(shù)據(jù)量上沒(méi)有公開(kāi)詳實(shí)的參考數(shù)據(jù)，而HBase早已在眾多實(shí)際項(xiàng)目特別是國(guó)內(nèi)一些知名互聯(lián)網(wǎng)公司中廣泛使用并得到了驗(yàn)證。另一方面，OpenTSDB和HBase都使用Java編寫(xiě)，這對(duì)于我們整個(gè)大數(shù)據(jù)技術(shù)團(tuán)隊(duì)來(lái)說(shuō)在維護(hù)和修復(fù)一些底層Bug上也相對(duì)容易一些。

TSD引擎驅(qū)動(dòng)

這是一個(gè)定制開(kāi)發(fā)的組件，其作用是對(duì)TSD數(shù)據(jù)進(jìn)行轉(zhuǎn)換和包裹，以便于更好地進(jìn)行數(shù)據(jù)展示，當(dāng)數(shù)據(jù)查詢(xún)請(qǐng)求到達(dá)時(shí)，它會(huì)根據(jù)請(qǐng)求的內(nèi)容和時(shí)間跨度把請(qǐng)求路由到實(shí)時(shí)庫(kù)或批處理庫(kù)，當(dāng)請(qǐng)求返回時(shí)，它同樣會(huì)過(guò)濾響應(yīng)內(nèi)容，對(duì)某些字段和值進(jìn)行映射和轉(zhuǎn)碼，如前所述，因?yàn)闀r(shí)間序列數(shù)據(jù)庫(kù)對(duì)存儲(chǔ)的TSD有很多形式上的限制，某些數(shù)據(jù)不可以直接存儲(chǔ)，它們?cè)谌霂?kù)前已經(jīng)做了相應(yīng)的格式化處理，在提取展示時(shí)需要進(jìn)行相應(yīng)的反處理。

TSD引擎驅(qū)動(dòng)本質(zhì)上是一個(gè)Web Service，從某種意義上說(shuō)，這個(gè)Web Service像是TSD引擎的一個(gè)反向代理，它能靈活和透明地解決一些定制化需求以及非標(biāo)準(zhǔn)數(shù)據(jù)的適配工作，從而避免對(duì)TSD引擎和前端展示進(jìn)行侵入性的修改。

在技術(shù)選型上，所有支持Web Service的框架都可以勝任這個(gè)工作，考慮到我們整個(gè)大平臺(tái)的技術(shù)堆棧都以sbt-native-packager/Java為主，我們實(shí)驗(yàn)性地選擇了Akka-Http，通過(guò)利用Akka-Http的HTTP DSL和sbt-native-packager的模式匹配，我們用很少的代碼就實(shí)現(xiàn)了既定目標(biāo)，效果非常好。

數(shù)據(jù)展示

最后，在數(shù)據(jù)展示上，Grafana是我們最佳的選擇。它是一個(gè)專(zhuān)門(mén)的時(shí)序數(shù)據(jù)展示工具，可以直連OpenTSDB，圖表的制作都是通過(guò)拖放完成的，它還有一個(gè)異常強(qiáng)大的“模版”機(jī)制，可以通過(guò)一次設(shè)定生成多張圖表。如果既有插件無(wú)法滿足展示需求，團(tuán)隊(duì)還以開(kāi)發(fā)自定義插件。

綜上所述，整個(gè)系統(tǒng)的技術(shù)堆棧如下所示：

物理架構(gòu)

對(duì)于平臺(tái)的物理架構(gòu)我們不打算進(jìn)行過(guò)多的介紹，因?yàn)镠adoop/Spark集群都大同小異，我們這里要討論的是這個(gè)平臺(tái)在物理架構(gòu)上的一個(gè)顯著的特點(diǎn)，就是我們構(gòu)建了兩個(gè)獨(dú)立的Hadoop/Spark集群，一個(gè)負(fù)責(zé)流處理，另一個(gè)負(fù)責(zé)批處理，這也是踐行Lambda架構(gòu)在物理層面上的一個(gè)自然的結(jié)果，兩個(gè)集群的數(shù)據(jù)交互依靠HBase的Replication機(jī)制透明地實(shí)現(xiàn)。其他的非Hadoop/Spark組件會(huì)部署在離散的服務(wù)器上。

實(shí)時(shí)處理集群和批處理集群除了分工上的不同，在集群結(jié)構(gòu)和節(jié)點(diǎn)配置上也有很大的區(qū)別，特別是在計(jì)算資源和存儲(chǔ)資源的分配上。通常，Hadoop集群的計(jì)算服務(wù)和存儲(chǔ)服務(wù)是共生在一起的，即HDFS的DataNode和YARN的NodeManager總是collocate的, 這樣做的目的是讓分布式計(jì)算盡可能地從本地讀取數(shù)據(jù)進(jìn)行處理，減少網(wǎng)絡(luò)IO，提升性能。我們的批處理集群就是按這樣的模式進(jìn)行資源配置的：基于Spark的批處理程序跑在Yarn的NodeManager上，盡量讀寫(xiě)本地DataNode上的數(shù)據(jù)，對(duì)于HBase也是同樣的邏輯，讓NodeManager也與DataNode共生在一起。

在實(shí)時(shí)處理集群上情況則大不相同。首先，在流處理過(guò)程中數(shù)據(jù)是不落地的，因此在流計(jì)算的節(jié)點(diǎn)上只會(huì)分配N(xiāo)odeManager，而不會(huì)有DataNode, 到了數(shù)據(jù)存儲(chǔ)環(huán)節(jié)才會(huì)讓HBase的NodeManager與DataNode共生。所以說(shuō)NodeManager和DataNode總是collocate的說(shuō)法太絕對(duì)，一切還是要根據(jù)實(shí)際情況靈活處理。

平臺(tái)建設(shè)

從前面介紹的技術(shù)架構(gòu)和選型上不難看出這個(gè)系統(tǒng)的復(fù)雜性，在建設(shè)過(guò)程中我們遇到了很多困難，也積累了一些寶貴的經(jīng)驗(yàn)，限于篇幅，我們選取了一些有價(jià)值的話題和大家進(jìn)行分享。

圍繞主數(shù)據(jù)進(jìn)行領(lǐng)域建模

“沒(méi)有領(lǐng)域模型的設(shè)計(jì)都是耍流氓”，這句看似調(diào)侃的話表達(dá)的卻是對(duì)軟件設(shè)計(jì)的一種嚴(yán)肅態(tài)度，領(lǐng)域模型在任何類(lèi)型的系統(tǒng)里都起著核心作用，大數(shù)據(jù)系統(tǒng)也不例外，你可以不去設(shè)計(jì)它，但這并不表示它不存在，一個(gè)不能如實(shí)反映業(yè)務(wù)邏輯的模型注定會(huì)導(dǎo)致整個(gè)系統(tǒng)的失敗。在我們這個(gè)面向時(shí)序數(shù)據(jù)的大數(shù)據(jù)平臺(tái)上，所有的TSD都出自于或描述了某一類(lèi)主數(shù)據(jù)的狀態(tài)或行為，或者說(shuō)它們都是主數(shù)據(jù)所代表的業(yè)務(wù)實(shí)體的產(chǎn)物，比如服務(wù)器的Metrics數(shù)據(jù)，這是典型的TSD，它們描述的就是業(yè)務(wù)對(duì)象：”服務(wù)器”的狀態(tài)。從OO建模的角度來(lái)思考這個(gè)問(wèn)題，如果監(jiān)控系統(tǒng)需要建立針對(duì)這個(gè)服務(wù)器的一整套監(jiān)控和報(bào)警規(guī)則，那么所有相應(yīng)的邏輯必然會(huì)追加到“服務(wù)器”以及一些和它相關(guān)聯(lián)的實(shí)體上，這就是我們所說(shuō)的“圍繞主數(shù)據(jù)進(jìn)行領(lǐng)域建?！?。

這一點(diǎn)非常重要且有效，因?yàn)樗菍?duì)所有業(yè)務(wù)邏輯的一種自然的梳理和劃分，最能夠反映領(lǐng)域的本來(lái)面目，越是復(fù)雜的業(yè)務(wù)場(chǎng)景越能體現(xiàn)優(yōu)越性。所有這些思考和傾向性都在引導(dǎo)我們漸漸地向“領(lǐng)域驅(qū)動(dòng)設(shè)計(jì)”(Domain Driven Design)的方向前進(jìn)，這是一個(gè)非常豐富并且具有實(shí)際意義的話題，令人感慨的是我們?cè)诖髷?shù)據(jù)平臺(tái)上讓“領(lǐng)域驅(qū)動(dòng)設(shè)計(jì)”再一次煥發(fā)了生機(jī)，以領(lǐng)域模型為核心驅(qū)動(dòng)業(yè)務(wù)處理和數(shù)據(jù)分析是一個(gè)非常明智的選擇，盡管這對(duì)團(tuán)隊(duì)整體的素質(zhì)有更高的要求，實(shí)施難度也更大，但是回饋也是巨大的。

我們有一個(gè)生動(dòng)的實(shí)例：在平臺(tái)建設(shè)的早期，限于每個(gè)數(shù)據(jù)源的格式和處理邏輯上的差異，每個(gè)數(shù)據(jù)源都自己的業(yè)務(wù)處理代碼和獨(dú)立的業(yè)務(wù)規(guī)則表，這種處理方式非常類(lèi)似于傳統(tǒng)企業(yè)應(yīng)用架構(gòu)中的“Transaction Script”模式(關(guān)于Transaction Script請(qǐng)參考Martin Fowler的《Patterns of Enterprise Application Architecture》一書(shū)的第9章)，伴隨著數(shù)據(jù)源的不斷引入，我們發(fā)現(xiàn)應(yīng)用在很多不同數(shù)據(jù)源上的監(jiān)控和報(bào)警邏輯都非常類(lèi)似，并且針對(duì)的業(yè)務(wù)對(duì)象也都是一樣的，例如不同的數(shù)據(jù)源都會(huì)面向某臺(tái)服務(wù)器或某個(gè)站點(diǎn)產(chǎn)生報(bào)警消息，而我們對(duì)這些報(bào)警消息的處理有著很大的相似性，這促使我們以主數(shù)據(jù)為對(duì)象進(jìn)行了領(lǐng)域建模，把邏輯進(jìn)行了統(tǒng)一梳理，在不一致的地方運(yùn)用適配器、修飾器和策略等模式進(jìn)行對(duì)接，最終將原來(lái)離散的代碼和配置統(tǒng)一在了一個(gè)領(lǐng)域模型上，大大簡(jiǎn)化了編程和維護(hù)成本，在處理新加入的數(shù)據(jù)源時(shí)變得更加簡(jiǎn)便快捷。

最后，補(bǔ)充一點(diǎn)認(rèn)識(shí)，在傳統(tǒng)企業(yè)級(jí)應(yīng)用里進(jìn)行領(lǐng)域驅(qū)動(dòng)設(shè)計(jì)有諸多的困難，其中一個(gè)比較突出的問(wèn)題就是領(lǐng)域?qū)ο蟮某志没捎跀?shù)據(jù)存放在關(guān)系型數(shù)據(jù)庫(kù)中，領(lǐng)域?qū)ο蟮膶?xiě)入和加載都存在一個(gè)“對(duì)象關(guān)系映射”的問(wèn)題，盡管有很多成熟的ORM框架能在一定程度上緩解這個(gè)問(wèn)題，但是在傳統(tǒng)企業(yè)級(jí)應(yīng)用里落地一個(gè)純正的領(lǐng)域模型依然是一個(gè)不小的挑戰(zhàn)，而大數(shù)據(jù)平臺(tái)為領(lǐng)域驅(qū)動(dòng)設(shè)計(jì)提供了一個(gè)更加自由的空間，比如大數(shù)據(jù)的計(jì)算節(jié)點(diǎn)可以提供足夠的內(nèi)存將領(lǐng)域?qū)ο笠淮涡匀考虞d，免去了ORM中對(duì)關(guān)聯(lián)對(duì)象加載策略的糾結(jié)，而領(lǐng)域?qū)ο髸?huì)在大數(shù)據(jù)處理過(guò)程中反復(fù)使用，客觀上也需要直接把它們加載到內(nèi)存中使用，再比如，在業(yè)務(wù)處理和分析階段，幾乎所有領(lǐng)域?qū)ο蠖际侵蛔x的，它們只會(huì)在同步主數(shù)據(jù)時(shí)被更新，這天然地形成了讀寫(xiě)分離，更加適合CQRS架構(gòu)。

流處理的工程結(jié)構(gòu)

很多團(tuán)隊(duì)在初次使用流計(jì)算框架構(gòu)建項(xiàng)目時(shí)往往會(huì)在如何組織工程結(jié)構(gòu)上感到迷茫，不同于傳統(tǒng)企業(yè)級(jí)應(yīng)用經(jīng)過(guò)多年積累形成的“套路”，流處理項(xiàng)目的工程結(jié)構(gòu)并沒(méi)有一個(gè)約定俗成的最佳實(shí)踐，我們?cè)谶@里分享我們的工程結(jié)構(gòu)作為一個(gè)參考，希望對(duì)你有所啟發(fā)。

也許你會(huì)覺(jué)得這個(gè)工程結(jié)構(gòu)非常面熟，是的，我們充分借鑒了傳統(tǒng)企業(yè)級(jí)應(yīng)用的分層結(jié)構(gòu)，每一個(gè)色塊都代表著一類(lèi)組件，映射到工程上就是一個(gè)package，讓我們逐一介紹一下：

Stream： 系統(tǒng)中的每一個(gè)流都會(huì)封裝在一個(gè)類(lèi)中，我們把這些類(lèi)統(tǒng)一按“XxxStream”形式進(jìn)行命名，放在stream包里，Stream類(lèi)里出現(xiàn)的多是與Spark Streaming相關(guān)的API，在涉及實(shí)際的業(yè)務(wù)處理時(shí)，會(huì)調(diào)用相應(yīng)的Service方法，這種設(shè)計(jì)反映了我們對(duì)流處理的一個(gè)基本認(rèn)識(shí)，那就是流計(jì)算中的API是一個(gè)“門(mén)面”(Facade)，厚重的業(yè)務(wù)處理不應(yīng)在這些API上直接以Lambda表達(dá)式的方式編寫(xiě)，而應(yīng)該封裝到專(zhuān)門(mén)的Service里。這與Web應(yīng)用中Action和Service的關(guān)系極為類(lèi)似。

Service： 與業(yè)務(wù)相關(guān)的處理邏輯會(huì)封裝到Service類(lèi)里，這是很傳統(tǒng)的做法，但是由于我們深度地應(yīng)用了領(lǐng)域驅(qū)動(dòng)設(shè)計(jì)，所以絕大部分的業(yè)務(wù)邏輯已經(jīng)自然地委派到了領(lǐng)域?qū)ο蟮姆椒ㄉ狭耍虼薙ervice也變成了很薄的一層封裝。有個(gè)值得一提的細(xì)節(jié)，我們把所有的Service都做成了object(sbt-native-packager中的object對(duì)象)，也就是單態(tài)， 這樣做的主要的動(dòng)機(jī)是讓所有的Executor節(jié)點(diǎn)在本地加載全局唯一的Service實(shí)例，避免Service實(shí)例從Driver端到Executor端做無(wú)謂的序列化與反序列化操作。

Repository：在相對(duì)簡(jiǎn)單的系統(tǒng)里，你可以利用Repository直接讀取存放于數(shù)據(jù)庫(kù)中的主數(shù)據(jù)和配置信息，如果你的平臺(tái)有多處組件都需要使用主數(shù)據(jù)，我們建議你務(wù)必建立統(tǒng)一的主數(shù)據(jù)和配置信息讀寫(xiě)組件，如果是這樣，則專(zhuān)屬于流處理的Repository將不復(fù)存在。

Domain：領(lǐng)域模型涉及的實(shí)體和值對(duì)象都會(huì)放在這個(gè)包里，業(yè)務(wù)處理和分析的邏輯會(huì)按照面向?qū)ο蟮脑O(shè)計(jì)理念分散到領(lǐng)域?qū)ο蟮臉I(yè)務(wù)方法上。同樣的，如果建立了統(tǒng)一的主數(shù)據(jù)和配置信息的讀寫(xiě)組件，則Domain也將不復(fù)存在

DTO： 流處理中的DTO并不是為傳輸領(lǐng)域?qū)ο蠖O(shè)計(jì)的，它是外部采集的原生數(shù)據(jù)經(jīng)過(guò)結(jié)構(gòu)化處理之后在流上的數(shù)據(jù)對(duì)象。

項(xiàng)目構(gòu)建：Sbt vs. Maven

由于我們的平臺(tái)技術(shù)堆棧以Spark為核心，我們的幾個(gè)核心組件都是使用scala編寫(xiě)的，在項(xiàng)目構(gòu)建上也積累了一些寶貴的經(jīng)驗(yàn)，早期我們使用的是scala的默認(rèn)構(gòu)建工具sbt， 作為新一代的構(gòu)建工具，sbt吸收了眾多前輩的優(yōu)點(diǎn)，簡(jiǎn)單易用，能夠滿足基本的應(yīng)用場(chǎng)景，但在實(shí)際的項(xiàng)目構(gòu)建中，當(dāng)面臨一些相對(duì)復(fù)雜的場(chǎng)景時(shí)，年青的sbt會(huì)顯得比較無(wú)力，其中最為我們不能接受的是面向多環(huán)境的構(gòu)建。盡管社區(qū)提出過(guò)一些解決方案，例如http://stackoverflow.com/questions/17193795/how-to-add-environment-profile-config-to-sbt ， 但是這個(gè)方案的缺陷在于對(duì)于每一套環(huán)境都要提供全套的配置，即使它們?cè)诙鄶?shù)據(jù)環(huán)境中的值是一樣的。實(shí)際上這個(gè)問(wèn)題的本質(zhì)原因是sbt尚沒(méi)有類(lèi)似Maven那樣在構(gòu)建時(shí)基于某個(gè)配置文件對(duì)一些變量進(jìn)行過(guò)濾和替換的Resource+Profile功能，這是很重要的一個(gè)需求。

在打包方面，我指的是構(gòu)建一個(gè)包含命令行工具、配置文件和各種lib的安裝包，sbt的sbt-native-packager確實(shí)非常強(qiáng)大，令人印象深刻。同樣，在面向不同環(huán)境的前提下，打包不同用途的package時(shí)，sbt-native-packager的靈活性還有待檢驗(yàn)。例如，基于我們過(guò)去的最佳實(shí)踐，面向每一種環(huán)境，我們嘗嘗會(huì)利用sbt-native-packager構(gòu)建兩種package，一種是包含全部產(chǎn)出物的標(biāo)準(zhǔn)部署包，一種是僅僅包含每次構(gòu)建都有可能發(fā)生變化的文件，例如項(xiàng)目自身的jar包和一些配置文件，我們把這種包稱(chēng)為最小化的package，這種package會(huì)用于日常持續(xù)集成的部署，它的體積很小，在網(wǎng)絡(luò)帶寬有限的環(huán)境里，它會(huì)大大節(jié)約部署時(shí)間。

回到Maven，在過(guò)去數(shù)年的開(kāi)發(fā)工作中，Maven滿足了我們各式各樣的構(gòu)建需求，從沒(méi)有讓我們失望過(guò)，它的約定大于配置的思想和豐富的周邊插件真正實(shí)踐了：”Make simple things simple， complex things possible!”從實(shí)際效果看，使用Maven構(gòu)建sbt-native-packager項(xiàng)目沒(méi)有任何障礙，它成熟而強(qiáng)大的各項(xiàng)功能可以解決實(shí)際項(xiàng)目上各式各樣的需求，這一切讓我們最終回歸了Maven。

但是這并不代表我們會(huì)在Maven上停滯不前，實(shí)際上我們對(duì)sbt依然抱有期望，只是它需要時(shí)間變得更加強(qiáng)大。在未來(lái)某個(gè)合適的時(shí)機(jī)，我想我們會(huì)遷移到sbt。

數(shù)據(jù)采集的痛點(diǎn)和應(yīng)對(duì)策略

數(shù)據(jù)采集往往是大數(shù)據(jù)平臺(tái)上的臟活、累活，除了解決技術(shù)上的問(wèn)題，團(tuán)隊(duì)還需要進(jìn)行大量協(xié)調(diào)和溝通工作，因?yàn)橥獠繑?shù)據(jù)源都由其他團(tuán)隊(duì)管理，需要從更高的組織層面進(jìn)行疏通，并且很多數(shù)據(jù)源需要同時(shí)為多個(gè)外部系統(tǒng)供給數(shù)據(jù)，為了確保數(shù)據(jù)源的可用性，會(huì)對(duì)外部的數(shù)據(jù)采集作業(yè)進(jìn)行控制，比如限制采集頻率等。我們下面會(huì)討論兩個(gè)棘手的問(wèn)題，并分享我們的解決方案。

數(shù)據(jù)采集作業(yè)超時(shí)

在我們采集的外部數(shù)據(jù)源中，有一個(gè)數(shù)據(jù)庫(kù)在某些時(shí)刻因?yàn)樾枰瑫r(shí)處理多個(gè)外圍系統(tǒng)疊加的查詢(xún)請(qǐng)求而經(jīng)常響應(yīng)緩慢，進(jìn)而導(dǎo)致了我們的數(shù)據(jù)采集作業(yè)超時(shí)，而這個(gè)Job原來(lái)的設(shè)計(jì)是每分鐘執(zhí)行一次，每次執(zhí)行時(shí)會(huì)從目標(biāo)數(shù)據(jù)庫(kù)中查詢(xún)最近1分鐘內(nèi)的數(shù)據(jù)，這個(gè)查詢(xún)請(qǐng)求通常在1秒以?xún)?nèi)就可以返回，但是當(dāng)數(shù)據(jù)庫(kù)響應(yīng)緩慢時(shí)，一個(gè)Job的耗時(shí)往往要超過(guò)1分鐘，而后續(xù)啟動(dòng)的Job仍然按啟動(dòng)時(shí)的時(shí)間點(diǎn)向前1分鐘作為時(shí)間窗口進(jìn)行查詢(xún)，這就出現(xiàn)了數(shù)據(jù)丟失。

應(yīng)對(duì)這個(gè)問(wèn)題的一個(gè)簡(jiǎn)單方案是將Job的執(zhí)行變?yōu)楫惒椒亲枞Ｊ?，每一個(gè)Job被觸發(fā)后都在一個(gè)獨(dú)立的線程中運(yùn)行。但是這個(gè)方案不適用于我們的系統(tǒng)，因?yàn)檫@樣采集到的數(shù)據(jù)不能保證時(shí)間上的有序性，而這對(duì)一個(gè)時(shí)序數(shù)據(jù)系統(tǒng)至關(guān)重要。所以這一方案被否決。

經(jīng)過(guò)仔細(xì)的思考，我們認(rèn)為必須要將這個(gè)Job切分成兩個(gè)子的Job：第一個(gè)Job負(fù)責(zé)制定周期性的計(jì)劃，準(zhǔn)確地說(shuō)是周期性地生成時(shí)間窗口參數(shù)，第二個(gè)Job負(fù)責(zé)讀取時(shí)間窗口參數(shù)執(zhí)行查詢(xún)，這一部分的工作并不是周期性的，原則上，只要有時(shí)間參數(shù)生成就應(yīng)該立即執(zhí)行，如果執(zhí)行超時(shí)，在超時(shí)期間，我們需要緩存第一個(gè)Job生成的時(shí)間參數(shù)，而當(dāng)所有的查詢(xún)都及時(shí)完成沒(méi)有待執(zhí)行的查詢(xún)計(jì)劃時(shí)，第二個(gè)Job需要等待新的查詢(xún)參數(shù)到達(dá)，是的，這實(shí)際上是一個(gè)生產(chǎn)者-消費(fèi)者模型，只是生產(chǎn)者是在“有節(jié)奏”地生產(chǎn)，在這個(gè)模式里，第三個(gè)參與者：倉(cāng)庫(kù)，或者說(shuō)傳送帶，起到了關(guān)鍵的調(diào)節(jié)作用，而一個(gè)現(xiàn)成的實(shí)現(xiàn)就是JDK自帶的BlockingQueue!于是我們的落地方案是：

第一個(gè)Job由定時(shí)器周期性觸發(fā)，每次觸發(fā)時(shí)會(huì)把當(dāng)前時(shí)間放到一個(gè)BlockingQueue的隊(duì)尾。

第二個(gè)Job循環(huán)執(zhí)行，每次執(zhí)行的工作就是從BlockingQueue的隊(duì)頭取出時(shí)間參數(shù)，組裝SQL并執(zhí)行。當(dāng)隊(duì)列為空時(shí)，由BlockingQueue來(lái)阻塞當(dāng)前線程，等待時(shí)間參數(shù)進(jìn)入隊(duì)列。

當(dāng)?shù)诙€(gè)Job執(zhí)行完一次時(shí)，如果隊(duì)列中還有時(shí)間參數(shù)，會(huì)繼續(xù)執(zhí)行步驟2，發(fā)生此類(lèi)情況時(shí)就說(shuō)明前一次的執(zhí)行超過(guò)了1分鐘。

數(shù)據(jù)延遲就緒

我們一直為降低平臺(tái)的數(shù)據(jù)延遲做著各種努力，但最讓人感到無(wú)力的是外部數(shù)據(jù)源本身在數(shù)據(jù)寫(xiě)入時(shí)發(fā)生了延遲。舉個(gè)例子，還是前面提到的數(shù)據(jù)庫(kù)，每次采集數(shù)據(jù)設(shè)定的時(shí)間區(qū)間是從當(dāng)前時(shí)間到前一分鐘，假定當(dāng)前時(shí)間是00:10，則執(zhí)行的SQL中時(shí)間窗口參數(shù)是(00:09，00:10]，此時(shí)你可能會(huì)查詢(xún)到1000條數(shù)據(jù)，但如果你在00:11以同樣的參數(shù)(00:09，00:10]再次執(zhí)行這條SQL， 返回的數(shù)據(jù)條目就可能變成了1200條，這說(shuō)明數(shù)據(jù)庫(kù)中的數(shù)據(jù)從它在業(yè)務(wù)系統(tǒng)中生成到最后寫(xiě)入數(shù)據(jù)庫(kù)的過(guò)程中發(fā)生了延遲，造成這種情況的原因有很多，比如系統(tǒng)存在性能問(wèn)題等等，總之現(xiàn)狀就是：數(shù)據(jù)就緒發(fā)生了延遲，而對(duì)于數(shù)據(jù)采集方這完全不可控。

面對(duì)這種問(wèn)題，我們的應(yīng)對(duì)策略是：如果數(shù)據(jù)及時(shí)地就緒了，我們要保證能及時(shí)的捕獲，如果數(shù)據(jù)延遲就緒，我們要保證至少不會(huì)丟掉它?；谶@樣的考慮，我們把同一個(gè)數(shù)據(jù)源的數(shù)據(jù)采集分成了兩到三個(gè)“波次”進(jìn)行，第一波次的采集緊緊貼近當(dāng)前時(shí)間，并且保持極高的頻率，這一波次是要保證最早最快地采集到當(dāng)前的新生數(shù)據(jù)，第二波次采集的是過(guò)去某個(gè)時(shí)間區(qū)間上的數(shù)據(jù)，時(shí)間偏移可能在十幾秒到幾分鐘不等，這取決于目標(biāo)數(shù)據(jù)源的數(shù)據(jù)延遲程度，第二波次是一個(gè)明顯的“補(bǔ)償”操作，用于采集在第一波次進(jìn)行時(shí)還未在數(shù)據(jù)庫(kù)中就緒的數(shù)據(jù)，第三波次則是最后的“托底”操作，它的時(shí)間偏移會(huì)更大，目的是最后一次補(bǔ)錄數(shù)據(jù)，保證數(shù)據(jù)的完整性。