呂鶴軒,黃 山,艾力卡木·再比布拉,吳思衡,段曉東

(1.大連民族大學(xué)計(jì)算機(jī)科學(xué)與工程學(xué)院，遼寧 大連 116600；2.大數(shù)據(jù)應(yīng)用技術(shù)國(guó)家民委重點(diǎn)實(shí)驗(yàn)室，遼寧 大連 116600； 3.大連市民族文化數(shù)字技術(shù)重點(diǎn)實(shí)驗(yàn)室，遼寧 大連 116600)

1 引言

近年來隨著信息技術(shù)的發(fā)展，各類門戶網(wǎng)站﹑搜索引擎和社交媒體軟件等產(chǎn)生的數(shù)據(jù)不斷膨脹，由互聯(lián)網(wǎng)數(shù)據(jù)中心IDC(Internet Data Center)發(fā)布的白皮書《數(shù)字化世界——從邊緣到核心》預(yù)測(cè)：到2025年，全球的數(shù)據(jù)量將增至175 ZB[1]。

大數(shù)據(jù)中有著非常豐富的數(shù)據(jù)信息，并且這些信息蘊(yùn)含著很高的分析和使用價(jià)值，這也就提高了對(duì)數(shù)據(jù)挖掘?qū)崟r(shí)性和準(zhǔn)確性的需求。隨著大數(shù)據(jù)時(shí)代的到來，傳統(tǒng)的單機(jī)處理框架無法滿足在巨大數(shù)據(jù)量的情況下的計(jì)算需求，分布式大數(shù)據(jù)處理框架應(yīng)運(yùn)而生。首先由谷歌提出的MapReduce[2]編程模型，到第1代大數(shù)據(jù)框架Hadoop[3]的誕生，到由內(nèi)存計(jì)算的批處理框架Spark[4]，再到現(xiàn)在具有高吞吐、低延遲等特性的流處理框架Flink[5]。大數(shù)據(jù)計(jì)算框架隨著大數(shù)據(jù)計(jì)算需求的提高不斷地發(fā)展。

衡量大數(shù)據(jù)的數(shù)據(jù)挖掘性能有2個(gè)最重要的任務(wù)指標(biāo)：一是實(shí)時(shí)性，如海量的數(shù)據(jù)規(guī)模需要實(shí)時(shí)分析并迅速反饋結(jié)果；二是準(zhǔn)確性，需要從海量的數(shù)據(jù)中精準(zhǔn)地提取出隱含在其中的用戶需要的有價(jià)值的信息。數(shù)據(jù)流的事件時(shí)間是數(shù)據(jù)流中事件實(shí)際發(fā)生的時(shí)間，他以附加在數(shù)據(jù)流中事件的時(shí)間戳為依據(jù)。事件時(shí)間將處理速度和結(jié)果徹底解耦，無論數(shù)據(jù)流的處理速度如何、事件時(shí)間到達(dá)算子的順序怎么樣，基于事件時(shí)間的窗口都會(huì)生成同樣的結(jié)果。

數(shù)據(jù)挖掘過程中通常需要窗口根據(jù)事件時(shí)間進(jìn)行聚合計(jì)算，而數(shù)據(jù)從產(chǎn)生到流入Kafka[6]消息隊(duì)列再經(jīng)過分布式處理框架數(shù)據(jù)源流入分布式處理框架進(jìn)行計(jì)算，往往因網(wǎng)絡(luò)傳輸速度不同、分布式節(jié)點(diǎn)計(jì)算性能不同等原因，導(dǎo)致數(shù)據(jù)流入算子的先后順序和數(shù)據(jù)事件時(shí)間存在著局部的亂序或者數(shù)據(jù)延遲現(xiàn)象。為解決此問題，F(xiàn)link提出了水位線機(jī)制。水位線是一個(gè)全局事件進(jìn)度指標(biāo)，通過設(shè)置最大允許亂序時(shí)間表示系統(tǒng)確信不會(huì)再有延遲事件到來的某個(gè)時(shí)間。窗口由事件的事件時(shí)間戳觸發(fā)開啟，窗口關(guān)閉和觸發(fā)計(jì)算則是通過水位線機(jī)制來主導(dǎo)。但是，水位線最大允亂序時(shí)間，也就是容錯(cuò)值的大小，使計(jì)算必須在實(shí)時(shí)性和準(zhǔn)確性之間進(jìn)行取舍。水位線設(shè)置過小會(huì)導(dǎo)致大部分?jǐn)?shù)據(jù)因?yàn)檫t到無法參與窗口計(jì)算，大大降低計(jì)算的準(zhǔn)確性；水位線設(shè)置過大時(shí)，雖然保證了大部分?jǐn)?shù)據(jù)都能參與計(jì)算，但是為等待太多嚴(yán)重遲到的無價(jià)值數(shù)據(jù)會(huì)導(dǎo)致窗口計(jì)算觸發(fā)延遲，過長(zhǎng)的等待時(shí)間大大降低了計(jì)算的實(shí)時(shí)性。綜上所述，在基于事件時(shí)間窗口的分布式計(jì)算中如何用更低的時(shí)延產(chǎn)生更少的遲到數(shù)據(jù)，使作業(yè)的窗口計(jì)算同時(shí)兼顧作業(yè)的準(zhǔn)確性和實(shí)時(shí)性，是一個(gè)急需解決的問題。

在不確定彈性數(shù)據(jù)流亂序程度的情況下，傳統(tǒng)的水位線設(shè)置效率低下，無法保證計(jì)算的實(shí)時(shí)性和準(zhǔn)確性。針對(duì)該問題，在不確定亂序程度的彈性流數(shù)據(jù)情況下，本文提出基于事件時(shí)間窗口的水位線動(dòng)態(tài)調(diào)整策略。該策略在不確定亂序程度的彈性流數(shù)據(jù)情況下，可以在保證大部分有價(jià)值數(shù)據(jù)計(jì)算準(zhǔn)確性的同時(shí)，有效提高計(jì)算的實(shí)時(shí)性。

本文的主要工作包括3個(gè)方面：

(1)提出了一種基于事件時(shí)間窗口的流數(shù)據(jù)微簇模型。該模型將事件時(shí)間的流數(shù)據(jù)按到達(dá)算子的順序分為微簇，此微簇的事件時(shí)間亂序程度代表當(dāng)前時(shí)刻局部數(shù)據(jù)流的事件時(shí)間亂序程度。

(2)提出了基于局部事件時(shí)間亂序度的水位線動(dòng)態(tài)調(diào)整策略。該策略根據(jù)局部數(shù)據(jù)的事件時(shí)間戳亂序程度動(dòng)態(tài)調(diào)整水位線大小，即允許最大亂序時(shí)間的大小。

(3)在Apache Flink框架上的對(duì)水位線動(dòng)態(tài)調(diào)整策略進(jìn)行實(shí)驗(yàn)。實(shí)驗(yàn)結(jié)果表明，相比傳統(tǒng)水位線機(jī)制，水位線動(dòng)態(tài)調(diào)整策略可以有效提高窗口計(jì)算的性能比。

2 相關(guān)工作

在現(xiàn)實(shí)世界中，基于事件時(shí)間的流處理系統(tǒng)因?yàn)榫W(wǎng)絡(luò)帶寬不同、節(jié)點(diǎn)性能不同等原因存在不確定彈性亂序問題，現(xiàn)有的水位線機(jī)制無法在保證計(jì)算的準(zhǔn)確性的同時(shí)保證計(jì)算的實(shí)時(shí)性。圖1所示為數(shù)據(jù)流在數(shù)據(jù)流時(shí)間模型的生命周期。

Figure 1 Life cycle of stream data 圖1 流數(shù)據(jù)生命周期

Shukla 等人[7]針對(duì)數(shù)據(jù)流不同的操作時(shí)間類型和統(tǒng)計(jì)的窗口技術(shù)進(jìn)行討論。Li等人[8]針對(duì)流數(shù)據(jù)亂序問題提出了一種新的核心流代數(shù)操作的物理實(shí)現(xiàn)策略，包括基于堆棧的數(shù)據(jù)結(jié)構(gòu)和相關(guān)的清除算法，采用物理的形式一定程度解決了數(shù)據(jù)流亂序問題，實(shí)驗(yàn)結(jié)果表明，物理策略可以有效解決流數(shù)據(jù)亂序問題。Bhatt等人[9]設(shè)計(jì)了一個(gè)可以處理流數(shù)據(jù)延遲的流處理模型，并提供了一個(gè)端到端的低延遲系統(tǒng)，通過實(shí)驗(yàn)表明，當(dāng)窗口大小等于數(shù)據(jù)到達(dá)率時(shí)，系統(tǒng)延遲可以有效減少。Affetti 等人[10]針對(duì)流處理的實(shí)時(shí)性以及窗口和時(shí)間的概念提出了Dataflow Model模型。Akidau等人[11]提出了針對(duì)時(shí)間語義進(jìn)行詳細(xì)檢測(cè)的數(shù)據(jù)流模型，解決了流數(shù)據(jù)無界和亂序的問題。Bhatt等人[12]通過分析水印和觸發(fā)方法來解決無界數(shù)據(jù)中的亂序和其他非常規(guī)問題。Bhatt等人[13]提出了在大數(shù)據(jù)處理中使用適當(dāng)?shù)墓艿篮退硖幚硌舆t和吞吐量的方法。高自娟等人[14]提出了一種基于變尺度滑動(dòng)窗口的流數(shù)據(jù)聚類算法。該算法采用動(dòng)態(tài)變化的滑動(dòng)窗口來采集流數(shù)據(jù)，利用帶有平均時(shí)間戳與平均權(quán)值的混合指數(shù)直方圖來支持?jǐn)?shù)據(jù)處理，從而能更好地捕獲動(dòng)態(tài)變化的流數(shù)據(jù)。徐江等人[15]結(jié)合基于分區(qū)與基于時(shí)間2種滑動(dòng)窗口思想，構(gòu)建單位時(shí)間周期下融合子流處理結(jié)果的滑動(dòng)窗口模型，采取并行實(shí)時(shí)運(yùn)算模式實(shí)現(xiàn)了實(shí)時(shí)流數(shù)據(jù)處理。

針對(duì)處理不確定彈性流數(shù)據(jù)亂序問題，雖然前人已經(jīng)做了相應(yīng)的研究并且獲得了一定的效果，但針對(duì)不確定亂序程度或彈性亂序流數(shù)據(jù)情況下的計(jì)算優(yōu)化研究仍有很大空間。

3 相關(guān)概念介紹

3.1 Flink框架

Flink是第3代流處理引擎，它支持精確的流處理，同時(shí)能滿足各種規(guī)模下對(duì)高吞吐和低延遲的要求。Flink對(duì)底層的操作進(jìn)行了封裝，從而為用戶提供了流處理Datastream API和批處理DataSet API 2個(gè)接口。用戶使用這些接口就可以完成基本的流數(shù)據(jù)處理任務(wù)和批數(shù)據(jù)處理任務(wù)。

3.2 Flink的流處理

3.2.1 Flink的時(shí)間語義

時(shí)間語義是Flink的四大基石之一。在Flink的流處理中會(huì)涉及到不同的時(shí)間概念，根據(jù)用戶不同事件類型需求，F(xiàn)link的時(shí)間主要分為事件時(shí)間、進(jìn)入時(shí)間和處理時(shí)間3類。事件時(shí)間是指事件創(chuàng)建的時(shí)間，它通常由事件所攜帶的時(shí)間戳表示，例如事件的生成時(shí)間。進(jìn)入時(shí)間是指數(shù)據(jù)流入Flink處理框架的時(shí)間。處理時(shí)間是指數(shù)據(jù)流入Flink框架后本節(jié)點(diǎn)對(duì)本條數(shù)據(jù)進(jìn)行計(jì)算操作的本地時(shí)間。

以上3種時(shí)間語義中，因?yàn)槭录r(shí)間將計(jì)算速度和計(jì)算結(jié)果內(nèi)容徹底解耦，無論數(shù)據(jù)流的處理速度如何、事件時(shí)間到達(dá)算子的順序怎么樣，基于事件時(shí)間的窗口都會(huì)生成同樣的結(jié)果，而且事件時(shí)間相對(duì)其他2種時(shí)間更具有研究?jī)r(jià)值，所以大多數(shù)流處理任務(wù)都會(huì)選擇使用數(shù)據(jù)的事件時(shí)間語義。

3.2.2 Flink的窗口分類

對(duì)于流處理系統(tǒng)而言，流入的消息不存在上限，處理的流數(shù)據(jù)可能是一個(gè)持續(xù)到達(dá)且無窮的事件流。所以，對(duì)于聚合操作和連接操作而言，流處理操作需要對(duì)流入的消息進(jìn)行分段處理，然后基于分段后的每一段消息進(jìn)行聚合計(jì)算或者連接等操作。此時(shí)的分段和分端口計(jì)算操作即為窗口。窗口是流式處理計(jì)算中一類十分常見的操作。Flink窗口操作主要分為滾動(dòng)窗口、滑動(dòng)窗口、會(huì)話窗口和全局窗口4類。滾動(dòng)窗口將無限的數(shù)據(jù)流按固定大小拆分成不同的窗口，不同的窗口之間的事件數(shù)據(jù)沒有交叉重疊。滑動(dòng)窗口有2個(gè)參數(shù)，分別是窗口大小和滑動(dòng)大小，不同窗口之間可以有事件數(shù)據(jù)交叉重疊。會(huì)話窗口用一個(gè)固定的時(shí)間間隔閾值來劃分不同的窗口。全局窗口把所有相同鍵值放入同一個(gè)窗口，全局窗口沒有起止的時(shí)間,需要自定義觸發(fā)計(jì)算，否則窗口永遠(yuǎn)不會(huì)進(jìn)行聚合計(jì)算。

3.2.3 基于事件時(shí)間的水位線

在使用事件時(shí)間處理流數(shù)據(jù)的時(shí)候會(huì)遇到數(shù)據(jù)亂序的問題，流處理從事件產(chǎn)生、流經(jīng)Source、再到操作算子需要一定的時(shí)間。理論情況下，傳輸?shù)讲僮魉阕拥臄?shù)據(jù)都是按照事件時(shí)間產(chǎn)生時(shí)的時(shí)間順序而來的，但是分布式環(huán)境下會(huì)因?yàn)榫W(wǎng)絡(luò)延遲、計(jì)算節(jié)點(diǎn)性能不同和數(shù)據(jù)背壓等原因而導(dǎo)致亂序的產(chǎn)生，特別是使用分布式消息隊(duì)列系統(tǒng)時(shí)，因此，大部分情況下多個(gè)分區(qū)之間的數(shù)據(jù)無法保證有序進(jìn)入窗口計(jì)算。圖2為數(shù)據(jù)流在系統(tǒng)中的理想順序和真實(shí)順序。但是，在進(jìn)行窗口計(jì)算的時(shí)候，不能無限期地等下去，必須要有一個(gè)機(jī)制來保證在特定的時(shí)間后觸發(fā)窗口計(jì)算。

Figure 2 Diagram of flow data out of order圖2 流數(shù)據(jù)亂序情況圖

水位線是一個(gè)進(jìn)度指標(biāo)，本質(zhì)上也是一種時(shí)間戳，根據(jù)水位線的插入類型分為有序水位線、無序水位線和多并行水位線。水位線表示系統(tǒng)確信不會(huì)再有延遲事件到來的某個(gè)時(shí)間點(diǎn)。水位線的工作原理是向數(shù)據(jù)流中插入當(dāng)前最大事件時(shí)間戳減去最大等待時(shí)間的時(shí)間戳，此時(shí)間戳即為水位線，計(jì)算窗口通過處理后的水位線來決定是否觸發(fā)窗口計(jì)算。等待時(shí)間也稱為容錯(cuò)值，即水位線的大小。例如，當(dāng)一個(gè)窗口算子接收到T的水位線，就可以認(rèn)為不會(huì)再有任何時(shí)間戳小于T的事件到來了。本質(zhì)上，水位線提供了一個(gè)邏輯時(shí)鐘，用來告知窗口當(dāng)前的事件時(shí)間，以及代替原來的數(shù)據(jù)事件時(shí)間觸發(fā)窗口關(guān)閉計(jì)算，但是窗口的開啟還是由數(shù)據(jù)的事件時(shí)間來觸發(fā)。圖3為水位線工作原理圖，系統(tǒng)按照規(guī)則向流數(shù)據(jù)中插入代表水位線的時(shí)間戳，當(dāng)水位線大小設(shè)為0 s時(shí)，水位線時(shí)間戳?xí)o挨著插入流數(shù)據(jù)，當(dāng)水位線大小設(shè)為2 s時(shí)，水位線時(shí)間戳?xí)跀?shù)據(jù)到達(dá)后等待2 s然后插入流數(shù)據(jù)中。

Figure 3 Principle of watermark圖3 水位線工作原理

水位線可用于平衡延遲和結(jié)果的完整性。較小的水位線保證了低延遲，但隨之而來的是低準(zhǔn)確性。該情況下，會(huì)有大部分有價(jià)值數(shù)據(jù)因?yàn)檫t到無法參與窗口計(jì)算。反之，如果水位線過大，雖然可信度得以保證，但可能會(huì)無謂地增加處理延遲。當(dāng)前的水位線雖然維持了低延遲和高準(zhǔn)確率的平衡，但是系統(tǒng)在面對(duì)未知亂序程度的數(shù)據(jù)流或者彈性亂序的數(shù)據(jù)流時(shí)無法做出彈性應(yīng)對(duì)。水位線只允許在結(jié)果的準(zhǔn)確性和延遲之間做出取舍，現(xiàn)有的方法并不能二者兼得。

4 基于事件時(shí)間窗口的動(dòng)態(tài)水位線

本節(jié)先對(duì)傳統(tǒng)水位線機(jī)制進(jìn)行簡(jiǎn)單分析，然后對(duì)基于事件時(shí)間窗口的流數(shù)據(jù)處理建立模型，之后介紹基于事件時(shí)間的局部亂序度算法的設(shè)計(jì)與實(shí)現(xiàn)，最后介紹針對(duì)事件時(shí)間窗口水位線動(dòng)態(tài)調(diào)整策略的設(shè)計(jì)與實(shí)現(xiàn)。

4.1 水位線設(shè)置大小分析

傳統(tǒng)的水位線常為根據(jù)流數(shù)據(jù)亂序情況和計(jì)算需求而設(shè)置的靜態(tài)值，但現(xiàn)實(shí)中多數(shù)流數(shù)據(jù)無法估計(jì)其亂序程度，而且大部分流數(shù)據(jù)內(nèi)部亂序情況也不是穩(wěn)定不變，而是會(huì)有局部彈性亂序的現(xiàn)象出現(xiàn)。

如圖4所示。白色空心圓表示遲到程度較小的數(shù)據(jù)，灰色實(shí)心圓表示遲到程度中等的數(shù)據(jù)，黑色實(shí)心圓表示遲到嚴(yán)重的數(shù)據(jù)。橫坐標(biāo)為事件時(shí)間,即數(shù)據(jù)產(chǎn)生時(shí)間;縱坐標(biāo)為到達(dá)窗口計(jì)算算子時(shí)間，理想水位線表示數(shù)據(jù)產(chǎn)生即實(shí)時(shí)到達(dá)，小水位線表示窗口關(guān)閉觸發(fā)計(jì)算等待較短時(shí)間，大水位線表示窗口關(guān)閉觸發(fā)計(jì)算等待較長(zhǎng)時(shí)間。靜態(tài)的水位線設(shè)置過小會(huì)導(dǎo)致大部分有價(jià)值的遲到不嚴(yán)重的數(shù)據(jù)無法參與窗口計(jì)算，從而降低窗口計(jì)算的準(zhǔn)確性。水位線設(shè)置過大，雖然保證了大部分遲到嚴(yán)重的數(shù)據(jù)也能參與窗口計(jì)算，但在流數(shù)據(jù)亂序情況穩(wěn)定或者不亂序的情況下延長(zhǎng)了窗口計(jì)算的等待時(shí)間，從而降低了窗口計(jì)算的實(shí)時(shí)性。綜上所述，需要一個(gè)可以根據(jù)數(shù)據(jù)流局部亂序情況動(dòng)態(tài)調(diào)整水位線大小的策略來應(yīng)對(duì)彈性亂序流數(shù)據(jù)，以保證水位線在流數(shù)據(jù)亂序嚴(yán)重的情況下升高使大部分有價(jià)值的數(shù)據(jù)也可以參與計(jì)算，水位線在流數(shù)據(jù)亂序穩(wěn)定的情況下降低，使窗口計(jì)算觸發(fā)不必等待過長(zhǎng)的時(shí)間。

Figure 4 Contrast diagram of watermark 圖4 水位線對(duì)比圖

4.2 基于事件時(shí)間窗口的流數(shù)據(jù)微簇模型

本節(jié)針對(duì)彈性亂序流數(shù)據(jù)情況下基于事件時(shí)間的窗口計(jì)算建立數(shù)據(jù)流微簇模型。首先對(duì)本文所用到的變量進(jìn)行定義。

ti：第i個(gè)到達(dá)數(shù)據(jù)所攜帶的事件時(shí)間戳。

N：當(dāng)前時(shí)間點(diǎn)所到達(dá)數(shù)據(jù)中的最大事件時(shí)間戳，如式(1)所示：

N=max(ti)

(1)

M：設(shè)置水位線時(shí)的最大等待時(shí)間，即水位線的動(dòng)態(tài)范圍。

Tk：當(dāng)數(shù)據(jù)流入計(jì)算數(shù)據(jù)流局部亂序度算子時(shí)，Tk表示當(dāng)前數(shù)據(jù)流微簇，如式(2)所示：

Tk={ti+1,ti+2,…，ti+k}

(2)

其中，k為微簇大小。

P：當(dāng)前時(shí)間微簇亂序程度，由微簇?cái)y帶信息求得。

Wi：插入到數(shù)據(jù)流中攜帶時(shí)間戳的水位線。當(dāng)前時(shí)間插入到數(shù)據(jù)流中的水位線Wi的計(jì)算如式(3)所示：

Wi=N-M*P

(3)

該模型針對(duì)事件時(shí)間窗口計(jì)算的水位線而動(dòng)態(tài)調(diào)整，主要提出微簇機(jī)制來表示當(dāng)前時(shí)刻流數(shù)據(jù)的亂序狀態(tài)，微簇Tk中按數(shù)據(jù)流入算子順序存儲(chǔ)并實(shí)時(shí)更新數(shù)據(jù)流到達(dá)當(dāng)前算子的最新k個(gè)時(shí)間戳，按照進(jìn)入算子的先后順序丟棄舊的時(shí)間戳，使微簇內(nèi)部始終保持k個(gè)最新時(shí)間戳。亂序數(shù)據(jù)多為數(shù)據(jù)背壓情況下各節(jié)點(diǎn)間網(wǎng)絡(luò)傳輸速度不同、各節(jié)點(diǎn)性能和計(jì)算速度不同所導(dǎo)致的，因此不能確定的彈性流數(shù)據(jù)具有亂序程度局部聚集特性[16]。即嚴(yán)重遲到的數(shù)據(jù)幾乎會(huì)聚集到同一時(shí)間段，同理亂序情況較弱的流數(shù)據(jù)也呈局部聚集狀態(tài)，因此微簇中部分最新數(shù)據(jù)的亂序情況可以有效地表示當(dāng)前時(shí)刻流數(shù)據(jù)的粗略亂序情況。使用該模型可以比較方便地求解。

4.3 基于事件時(shí)間的局部亂序度算法

以建立最優(yōu)水位線調(diào)整模型為基礎(chǔ)，可以設(shè)計(jì)出基于事件時(shí)間的局部亂序度算法,具體如算法1所示。

算法1基于事件時(shí)間的局部亂序度算法

輸入：當(dāng)前最新微簇Tk，微簇Tk攜帶當(dāng)前最新k個(gè)已到達(dá)數(shù)據(jù)所攜帶時(shí)間戳信息和到達(dá)先后順序信息。

輸出：0～1的亂序度P。

Step1根據(jù)數(shù)據(jù)流微簇模型，算子內(nèi)部創(chuàng)建狀態(tài)存放長(zhǎng)度為k的微簇Tk、當(dāng)前亂序度P及當(dāng)前數(shù)據(jù)流最大事件時(shí)間戳，微簇按到達(dá)算子先后順序存儲(chǔ)當(dāng)前最新k個(gè)到達(dá)算子的數(shù)據(jù)事件時(shí)間戳。

Step2當(dāng)有新的數(shù)據(jù)到達(dá)當(dāng)前算子時(shí)更新微簇Tk內(nèi)部事件時(shí)間戳和算子最大事件時(shí)間戳N，使微簇始終保持按到達(dá)先后順序存儲(chǔ)最新事件時(shí)間戳,如圖5所示為微簇狀態(tài)更新圖。

Figure 5 Updating of microcluster state 圖5 微簇狀態(tài)更新圖

Step3當(dāng)微簇更新狀態(tài)時(shí)，計(jì)算亂序度P，亂序度計(jì)算方式如式(4)所示：

(4)

其中，P表示當(dāng)前微簇的亂序程度，其值在0～1，1表示微簇完全正序，0表示微簇完全逆序。

Step4更新亂序度P，即代表當(dāng)前數(shù)據(jù)流局部亂序程度。

該算法根據(jù)流數(shù)據(jù)微簇模型中微簇的狀態(tài)實(shí)時(shí)計(jì)算當(dāng)前數(shù)據(jù)流的亂序度P。亂序度P由數(shù)據(jù)進(jìn)入微簇的先后順序和數(shù)據(jù)所攜帶的時(shí)間戳決定，亂序度越大表示當(dāng)前微簇亂序程度越低，亂序度越小表示當(dāng)前微簇亂序程度越高。

4.4 水位線動(dòng)態(tài)調(diào)整策略的設(shè)計(jì)與實(shí)現(xiàn)

在并行集群中，流數(shù)據(jù)從數(shù)據(jù)源流入系統(tǒng)將根據(jù)編程邏輯結(jié)構(gòu)被分配到不同節(jié)點(diǎn)的不同算子上，但因?yàn)橛?jì)算資源不同和網(wǎng)絡(luò)傳輸速度不同等原因會(huì)導(dǎo)致數(shù)據(jù)事件時(shí)間局部亂序或延遲等情況發(fā)生。為保證計(jì)算任務(wù)高效率完成，需要更準(zhǔn)確的策略來調(diào)整水位線容錯(cuò)值。本文提出一種基于事件時(shí)間局部亂序度算法的水位線動(dòng)態(tài)調(diào)整策略，其流程圖如圖6所示。

Figure 6 Flowchart of dynamic watermark adjustment strategy圖6 水位線動(dòng)態(tài)調(diào)整策略流程

具體步驟為：

(1)用戶提交作業(yè)后，計(jì)算引擎會(huì)根據(jù)用戶定義的數(shù)據(jù)源將數(shù)據(jù)流讀取到分布式環(huán)境之中，并將數(shù)據(jù)按照編程的邏輯傳遞到各個(gè)節(jié)點(diǎn)上。

(2)流數(shù)據(jù)從數(shù)據(jù)源流入系統(tǒng)以后，系統(tǒng)內(nèi)部會(huì)根據(jù)處理數(shù)據(jù)的內(nèi)部邏輯對(duì)流數(shù)據(jù)進(jìn)行分區(qū)、計(jì)算和處理。

(3)為不同并行度上的算子分配算子狀態(tài)。該方法基于數(shù)據(jù)流微簇模型將當(dāng)前算子流中的數(shù)據(jù)分為微簇，微簇時(shí)間戳數(shù)據(jù)存儲(chǔ)在當(dāng)前算子狀態(tài)中。

(4)亂序度計(jì)算。當(dāng)有新的數(shù)據(jù)流入算子時(shí)，根據(jù)微簇中的時(shí)間戳和基于事件時(shí)間的局部亂序度算法計(jì)算當(dāng)前時(shí)刻的亂序度。

(5)當(dāng)有新的數(shù)據(jù)流流入算子時(shí)，算子內(nèi)部首先根據(jù)數(shù)據(jù)事件時(shí)間更新微簇內(nèi)時(shí)間戳狀態(tài)，然后根據(jù)微簇計(jì)算當(dāng)前時(shí)間的局部亂序度。

(6)更新水位線最大允許亂序時(shí)間值。當(dāng)計(jì)算出新的局部亂序度以后，根據(jù)局部亂序度大小調(diào)整水位線最大允許亂序時(shí)間值。調(diào)整方式如式(5)所示：

Wi=N-M*P

(5)

(7)更新當(dāng)前最大水位線Wi，確保插入數(shù)據(jù)流中的水位線呈正序增加狀態(tài)。

(8)當(dāng)?shù)玫叫碌乃痪€以后，按照程序中設(shè)定的水位線周期將水位線插入數(shù)據(jù)流中并傳遞給下一個(gè)算子。

(9)當(dāng)事件時(shí)間窗口收到大于窗口關(guān)閉的水位線值后觸發(fā)窗口關(guān)閉并執(zhí)行程序。

如圖7所示，當(dāng)水位線的允許亂序時(shí)間設(shè)置過大，會(huì)出現(xiàn)窗口無效等待的情況，從而降低了系統(tǒng)的計(jì)算速度；當(dāng)水位線的允許亂序時(shí)間設(shè)置過小會(huì)使窗口丟失數(shù)據(jù)影響窗口計(jì)算的準(zhǔn)確性。而本策略可以使最大允許等待時(shí)間動(dòng)態(tài)變化，在保證系統(tǒng)計(jì)算準(zhǔn)確性的同時(shí)降低計(jì)算延遲。

Figure 7 Window function圖7 窗口函數(shù)

5 實(shí)驗(yàn)設(shè)計(jì)與結(jié)果分析

本文提出的水位線動(dòng)態(tài)調(diào)整策略在Apache Flink 1.12中進(jìn)行了實(shí)現(xiàn)，本節(jié)介紹實(shí)驗(yàn)環(huán)境和實(shí)驗(yàn)形式并根據(jù)實(shí)驗(yàn)結(jié)果進(jìn)行分析，以證明該策略的有效性。

5.1 實(shí)驗(yàn)環(huán)境與實(shí)驗(yàn)設(shè)計(jì)

5.1.1 實(shí)驗(yàn)環(huán)境

實(shí)驗(yàn)環(huán)境是由4臺(tái)機(jī)器搭建的Flink集群，集群具體參數(shù)配置如表1所示。集群設(shè)置1臺(tái)主節(jié)點(diǎn)Master，3臺(tái)從節(jié)點(diǎn)Node1、Node2和Node3。集群環(huán)境為CentOS 7.4.1系統(tǒng)，Java 1.8.0 版本。并使用Java語言進(jìn)行程序編寫。

Table 1 Configuration of flink cluster 表1 Flink集群配置

Flink集群支持Standalone cluster、Flink on Yarn和Flink on Kubernetes 3種模式，本實(shí)驗(yàn)采用常用的Standalone cluster模式進(jìn)行集群部署。

5.1.2 實(shí)驗(yàn)設(shè)計(jì)

因?yàn)楸疚乃岢龅牟呗灾饕轻槍?duì)窗口計(jì)算處理不同情況所導(dǎo)致的局部亂序流任務(wù)，所以數(shù)據(jù)集采用模擬生成的彈性亂序數(shù)據(jù)集，數(shù)據(jù)強(qiáng)度分別設(shè)置為10萬，20萬，50萬，80萬，100萬和150萬條流數(shù)據(jù)作為測(cè)試數(shù)據(jù)集，每次實(shí)驗(yàn)用盡數(shù)據(jù)集。為驗(yàn)證該策略和亂序度算法消耗的系統(tǒng)資源較小且在任務(wù)中可忽略不計(jì)，特設(shè)置每1 ms生成一條數(shù)據(jù)，根據(jù)馬卿云等人[17]對(duì)同等級(jí)集群帶寬傳輸開銷測(cè)試，數(shù)據(jù)亂序程度分別采用3 s, 5 s, 8 s, 10 s和12 s延遲上下浮動(dòng)。算法亂序度分別采用3 s, 5 s, 8 s, 10 s和12 s進(jìn)行對(duì)比。實(shí)驗(yàn)采用流處理數(shù)據(jù)密集型任務(wù)WordCount作為窗口任務(wù)，窗口滑動(dòng)距離20 s。并行度分別設(shè)為1,4,8,12,20和32。

5.2 實(shí)驗(yàn)評(píng)估標(biāo)準(zhǔn)

因?yàn)楸疚脑O(shè)計(jì)的實(shí)驗(yàn)為窗口任務(wù)處理不同亂序程度的流數(shù)據(jù)，故主要選取性能比作為評(píng)估標(biāo)準(zhǔn),從有效性、魯棒性、可擴(kuò)展性和抗壓性4個(gè)方面進(jìn)行對(duì)比實(shí)驗(yàn)。

性能比根據(jù)式(6)計(jì)算：

(6)

其中,Ni為參與窗口計(jì)算的數(shù)據(jù)量，Ni越大即計(jì)算完整沒有遲到數(shù)據(jù)或者低價(jià)值遲到數(shù)據(jù)越少，說明窗口計(jì)算準(zhǔn)確性高；Di為窗口從開始到觸發(fā)計(jì)算結(jié)束所用時(shí)長(zhǎng)，即窗口計(jì)算時(shí)延，Di越小說明窗口計(jì)算時(shí)延越低,即實(shí)時(shí)性越高。m為開啟窗口個(gè)數(shù)。性能比E即為平均計(jì)算數(shù)據(jù)量比平均窗口計(jì)算時(shí)長(zhǎng)。性能比越高,說明計(jì)算兼顧準(zhǔn)確性和實(shí)時(shí)性的能力越強(qiáng)。

5.3 實(shí)驗(yàn)結(jié)果與分析

5.3.1 有效性測(cè)評(píng)

本節(jié)分別為動(dòng)態(tài)水位線算法和傳統(tǒng)固定水位線算法設(shè)置不同的最大等待時(shí)間，實(shí)驗(yàn)結(jié)果如表2所示。

Table 2 Performance comparison of watermark algorithm with different algorithms out-of-order degrees表2 水位線算法在不同算法亂序度下性能對(duì)比

從表2中可以看出，最初最大等待時(shí)間太小限制了動(dòng)態(tài)水位線算法的動(dòng)態(tài)范圍，所以性能比略差于傳統(tǒng)固定水位線算法的；但是，隨著最大等待時(shí)間設(shè)置的增大，動(dòng)態(tài)水位線算法動(dòng)態(tài)范圍變大，可以更靈活地處理亂序數(shù)據(jù)，性能比也慢慢高于傳統(tǒng)固定水位線算法算法的。

5.3.2 魯棒性測(cè)評(píng)

本節(jié)分別為動(dòng)態(tài)水位線算法和傳統(tǒng)固定水位線算法設(shè)置不同程度的亂序數(shù)據(jù)，實(shí)驗(yàn)結(jié)果如表3所示。從表3中可以看出，面對(duì)不同亂序程度的數(shù)據(jù)流，動(dòng)態(tài)水位線算法性能比均高于傳統(tǒng)固定水位線算法的。但是，當(dāng)數(shù)據(jù)流亂序程度高于動(dòng)態(tài)水位線算法最大等待時(shí)間范圍時(shí)，性能比會(huì)略微下降。這是由于不論動(dòng)態(tài)水位線多靈活，也是有界限范圍的，超出動(dòng)態(tài)范圍的嚴(yán)重遲到數(shù)據(jù)還是無法進(jìn)入窗口計(jì)算。延遲超大的不確定彈性數(shù)據(jù)還是無法保證全部進(jìn)入窗口計(jì)算。為了均衡實(shí)時(shí)性和準(zhǔn)確性，有時(shí)不得不舍棄一些嚴(yán)重遲到的數(shù)據(jù)。而且，嚴(yán)重遲到的數(shù)據(jù)往往價(jià)值不大，大部分情況下舍去低價(jià)值數(shù)據(jù)避免過高的時(shí)延是非常有必要的。

Table 3 Performance comparison of watermark algorithm with different data out-of-order degrees表3 水位線算法在不同數(shù)據(jù)亂序度下性能對(duì)比

5.3.3 可擴(kuò)展性測(cè)評(píng)

本節(jié)分別為動(dòng)態(tài)水位線算法和傳統(tǒng)固定水位線算法設(shè)置不同的并行度，其中模擬數(shù)據(jù)亂序程度設(shè)為8 s，算法最大等待時(shí)間設(shè)為8 s。實(shí)驗(yàn)結(jié)果如表4所示。從表4中可以看出，在不同并行度的情況下，動(dòng)態(tài)水位線算法性能比上下浮動(dòng)小，且均優(yōu)于傳統(tǒng)固定水位線算法的。這說明動(dòng)態(tài)水位線算法和調(diào)整策略效果穩(wěn)定，具有一定的可擴(kuò)展性。

Table 4 Performance comparison of watermark algorithms with different parallelisms表4 水位線算法在不同并行度下性能對(duì)比

5.3.4 抗壓性測(cè)評(píng)

本文分別為動(dòng)態(tài)水位線算法和傳統(tǒng)固定水位線算法設(shè)置不同強(qiáng)度的數(shù)據(jù)量，其中模擬數(shù)據(jù)亂序程度設(shè)為8 s，算法最大等待時(shí)間設(shè)為8 s。因不同數(shù)據(jù)量和固定的窗口長(zhǎng)度導(dǎo)致性能比無法對(duì)比，所以性能比按數(shù)據(jù)量比例采用平均每10萬條數(shù)據(jù)量進(jìn)行計(jì)算。實(shí)驗(yàn)結(jié)果如表5所示。從表5中可以看出，面對(duì)不同強(qiáng)度的數(shù)據(jù)量，動(dòng)態(tài)水位線算法的性能比均高于傳統(tǒng)固定水位線算法的，且性能穩(wěn)定。這說明動(dòng)態(tài)水位線算法和調(diào)整策略在現(xiàn)有強(qiáng)度測(cè)試下效果穩(wěn)定，具有一定的抗壓性。

Table 5 Performance comparison of watermark algorithms with different data sizes表5 水位線算法在不同數(shù)據(jù)量下性能對(duì)比

6 結(jié)束語

本文針對(duì)傳統(tǒng)固定水位線算法針對(duì)基于事件時(shí)間的不確定彈性亂序流數(shù)據(jù)的窗口計(jì)算無法同時(shí)兼顧準(zhǔn)確性和實(shí)時(shí)性的問題，提出了基于局部亂序度算法的水位線動(dòng)態(tài)調(diào)整策略。該策略基于流數(shù)據(jù)時(shí)間模型和局部亂序度算法對(duì)水位線進(jìn)行動(dòng)態(tài)調(diào)整。通過模擬流數(shù)據(jù)進(jìn)行實(shí)驗(yàn)測(cè)評(píng)，分析對(duì)比了2種水位線算法的準(zhǔn)確性和平均時(shí)長(zhǎng)。實(shí)驗(yàn)結(jié)果表明，系統(tǒng)在保證大部分?jǐn)?shù)據(jù)不丟失的前提下，縮短了窗口等待時(shí)間，提高了窗口計(jì)算的實(shí)時(shí)性，為因計(jì)算資源不同和網(wǎng)絡(luò)傳輸速率不同等原因?qū)е碌臄?shù)據(jù)彈性亂序情況下的窗口計(jì)算提供了可靠的保證。

未來將考慮在異構(gòu)集群環(huán)境中不同節(jié)點(diǎn)計(jì)算性能差距明顯的情況下進(jìn)一步優(yōu)化流計(jì)算模型，并考慮在更大的數(shù)據(jù)集上進(jìn)行實(shí)驗(yàn)驗(yàn)證，以及在真實(shí)場(chǎng)景下進(jìn)行應(yīng)用，以測(cè)試系統(tǒng)平臺(tái)效果，并優(yōu)化流計(jì)算算法。