Abstract

彈性非常適用于流系統(tǒng)，以保證針對(duì)工作負(fù)載動(dòng)態(tài)的低延遲，例如到達(dá)率的激增和數(shù)據(jù)分布的波動(dòng)。現(xiàn)有系統(tǒng)使用以resource-centric的方法實(shí)現(xiàn)彈性，該方法在并行實(shí)例（即執(zhí)行程序）之間重新分配Key，以平衡工作負(fù)載和擴(kuò)展Operator。然而，這種Operator級(jí)別的重新分區(qū)需要全局同步并且禁止快速?gòu)椥浴Ｎ覀兲岢隽艘环N以executor-centric的方法，它避免了Operator級(jí)別的Key數(shù)據(jù)重新分區(qū)，并將執(zhí)行程序作為彈性的構(gòu)建塊。通過(guò)這種新方法，我們?cè)O(shè)計(jì)了具有兩級(jí)優(yōu)化的Elasticutor框架：i）執(zhí)行器的新穎實(shí)現(xiàn)，即elastic executors，通過(guò)有效的executror內(nèi)負(fù)載均衡和executor擴(kuò)展來(lái)執(zhí)行彈性多核執(zhí)行，以及ii）a基于全局模型的調(diào)度程序，可根據(jù)瞬時(shí)工作負(fù)載為執(zhí)行程序動(dòng)態(tài)分配CPU內(nèi)核。我們實(shí)施了一個(gè)原型Elasticutor并進(jìn)行了大量實(shí)驗(yàn)。我們表明，與實(shí)際應(yīng)用程序的動(dòng)態(tài)工作負(fù)載方法相比，Elasticutor的吞吐量增加了一倍，延遲降低了兩個(gè)數(shù)量級(jí)。

1. Introduction

?????? 分布式流系統(tǒng)[8,12,40,43,45,50,51]實(shí)現(xiàn)了對(duì)連續(xù)流的實(shí)時(shí)數(shù)據(jù)處理，并已廣泛用于欺詐檢測(cè)，監(jiān)控分析和定量融資等應(yīng)用。 在這樣的系統(tǒng)中，應(yīng)用程序邏輯被建模為計(jì)算圖，其中每個(gè)頂點(diǎn)表示與用戶(hù)定義的處理邏輯相關(guān)聯(lián)的Operator，并且每個(gè)邊指定運(yùn)算符之間的數(shù)據(jù)流的輸入-輸出關(guān)系。 為了實(shí)現(xiàn)大規(guī)模數(shù)據(jù)處理，輸入數(shù)據(jù)流通常被定義為通過(guò)key 空間分區(qū)到下游子分區(qū)中。并行執(zhí)行實(shí)例（即執(zhí)行程序）以將每個(gè)key的子空間靜態(tài)綁定到一定量的計(jì)算資源，通常是CPU核心。 結(jié)果，每個(gè)執(zhí)行器可以獨(dú)立地進(jìn)行與其key SubSpace相關(guān)聯(lián)的計(jì)算。

?????? 然而，在股票交易和視頻分析等實(shí)際應(yīng)用中，工作量隨時(shí)間波動(dòng)很大，導(dǎo)致嚴(yán)重的性能下降[15,39]。 從時(shí)間角度來(lái)看，發(fā)送到Operator的總工作量可能會(huì)在短時(shí)間內(nèi)顯著激增，例如10秒，這使得Operator成為整個(gè)處理流程的瓶頸。 從空間角度來(lái)看，Key空間上的工作負(fù)載分布可能不穩(wěn)定，導(dǎo)致執(zhí)行程序中的工作負(fù)載偏差，其中一些CPU利用率較低，而另一些則過(guò)載。 為了適應(yīng)工作負(fù)載波動(dòng)，先前的工作[14,15,39,41]提出了實(shí)現(xiàn)彈性的解決方案，即Operator擴(kuò)展和負(fù)載平衡。 所有這些解決方案都以資源為中心，因?yàn)閳?zhí)行程序受特定資源的約束，并且通過(guò)跨執(zhí)行程序動(dòng)態(tài)地重新分配Key來(lái)實(shí)現(xiàn)彈性。

?????? 圖1（a）說(shuō)明了由于工作負(fù)載分配不平衡而導(dǎo)致執(zhí)行程序過(guò)載的情況。為了減輕性能瓶頸，重新分區(qū)Key空間，以便將重載執(zhí)行程序中的一定數(shù)量的工作負(fù)載與相應(yīng)的Key一起遷移到負(fù)載較輕的執(zhí)行程序。但是，這個(gè)過(guò)程需要一個(gè)耗時(shí)的協(xié)議[15,39]來(lái)維持狀態(tài)一致性。特別是，系統(tǒng)需要執(zhí)行以下操作：（a）阻止上游執(zhí)行程序向下游發(fā)送元組; （b）等待所有飛行中的元組進(jìn)行處理; （c）根據(jù)新的Key空間劃分，將狀態(tài)遷移到新的key space分區(qū)匯總; （d）更新上游執(zhí)行者的路由表;最后（e）恢復(fù)向下游發(fā)送元組的上游執(zhí)行者。由于Operator間路由更新和執(zhí)行器間狀態(tài)遷移都需要昂貴的全局同步，因此Key Space重新分配可能持續(xù)數(shù)秒，在此期間無(wú)法處理新的傳入元組并導(dǎo)致嚴(yán)重延遲。

?????? 為了實(shí)現(xiàn)快速?gòu)椥?#xff0c;我們提出了一種以執(zhí)行者為中心的范式。 核心思想是在執(zhí)行程序之間靜態(tài)劃分Operator的Key Space，但根據(jù)其瞬時(shí)工作負(fù)載動(dòng)態(tài)地為每個(gè)執(zhí)行程序分配CPU核心。 圖1（b）說(shuō)明了新方法不是對(duì)Key Space進(jìn)行重新分區(qū)，而是通過(guò)將CPU內(nèi)核從較輕負(fù)載的執(zhí)行程序重新分配給過(guò)載的執(zhí)行程序來(lái)平衡工作負(fù)載。 由于每個(gè)執(zhí)行器擁有固定Key SubSpace，新方法實(shí)現(xiàn)了Operator間獨(dú)立性，即上游Operator不需要與下游Operator同步，并且執(zhí)行器間獨(dú)立性，即與Key SubSpace相關(guān)聯(lián)的狀態(tài)不需要遷移跨執(zhí)行者。 換句話(huà)說(shuō)，這種新方法優(yōu)雅地解耦了Operator Key Space重新分區(qū)和計(jì)算資源的動(dòng)態(tài)供應(yīng)之間的綁定。

?????? 基于以執(zhí)行器為中心的方法，我們?cè)O(shè)計(jì)了具有兩個(gè)優(yōu)化級(jí)別的Elasticutor框架。 在執(zhí)行程序級(jí)別，作為輕量級(jí)分布式子系統(tǒng)實(shí)現(xiàn)，每個(gè)彈性執(zhí)行程序在其分配的CPU核心上均勻分配其工作負(fù)載，并在調(diào)度程序分配/取消分配CPU核心時(shí)快速擴(kuò)展。 在整體層面，基于模型的動(dòng)態(tài)調(diào)度程序設(shè)計(jì)用于根據(jù)測(cè)量的性能指標(biāo)Metrics優(yōu)化核心到執(zhí)行程序的分配，以便以最小的狀態(tài)遷移開(kāi)銷(xiāo)和最大的計(jì)算局部性來(lái)適應(yīng)工作負(fù)載動(dòng)態(tài)。 我們實(shí)現(xiàn)了Elasticutor的原型，并使用合成和真實(shí)數(shù)據(jù)集進(jìn)行了大量的實(shí)驗(yàn)。 結(jié)果表明，Elasticutor使吞吐量翻倍，并且比現(xiàn)有方法實(shí)現(xiàn)了更低的延遲。

?????? 本文的其余部分安排如下。 第2節(jié)介紹了以執(zhí)行者為中心的范例，并概述了Elasticutor框架。 第3節(jié)和第4節(jié)分別介紹了彈性執(zhí)行器和動(dòng)態(tài)調(diào)度器的設(shè)計(jì)。 第5節(jié)討論了實(shí)驗(yàn)結(jié)果。 第6節(jié)回顧了相關(guān)工作。 第7節(jié)總結(jié)了論文。

2. PARADIGM AND FRAMEWORK

2.1 Basic Concepts

?????? 我們考慮在由快速網(wǎng)絡(luò)設(shè)備連接的稱(chēng)為節(jié)點(diǎn)的機(jī)器群集上的實(shí)時(shí)有狀態(tài)流處理系統(tǒng)。流是一個(gè)無(wú)限的序列。來(lái)自輸入流的元組連續(xù)到達(dá)系統(tǒng)并立即處理。用戶(hù)應(yīng)用程序被建模為計(jì)算的有向圖，稱(chēng)為T(mén)opology，其中頂點(diǎn)是具有用戶(hù)定義的處理邏輯的運(yùn)算符，并且邊表示運(yùn)算符之間的處理序列。對(duì)于每對(duì)相鄰Operator，流的元組由上游Operator生成并由下游Operator消費(fèi)。在有狀態(tài)計(jì)算中，Operator維護(hù)內(nèi)部狀態(tài)，該狀態(tài)用于計(jì)算并將在輸入元組的處理期間更新。為了分配和并行化計(jì)算，操作符的狀態(tài)被實(shí)現(xiàn)為在key space上定義的可分割數(shù)據(jù)結(jié)構(gòu)。系統(tǒng)將key space劃分為sub space，并創(chuàng)建一個(gè)稱(chēng)為執(zhí)行程序的并行實(shí)例，每個(gè)實(shí)例具有相同的數(shù)據(jù)處理邏輯。為了保證在這樣的分布式系統(tǒng)上維護(hù)的狀態(tài)的一致性，需要將元組正確路由到下游執(zhí)行器。因?yàn)橐圆煌樞蛱幚硐嗤妮斎朐M序列可能導(dǎo)致不同的輸出元組和狀態(tài)，所以有狀態(tài)計(jì)算中的另一個(gè)基本要求是按到達(dá)順序處理相同key的元組。

?????? 流處理工作負(fù)載通常是動(dòng)態(tài)的，因?yàn)閷?duì)Operator的輸入速率和元組的key分配隨時(shí)間波動(dòng)。為了保證動(dòng)態(tài)工作負(fù)載下的性能，應(yīng)該向Operator適當(dāng)?shù)靥峁┯?jì)算資源，即CPU核心，以便確保1）Operator擴(kuò)展，即CPU核心根據(jù)其工作負(fù)載動(dòng)態(tài)地分配給Operator; 2）負(fù)載平衡，即每個(gè)Operator的工作負(fù)載均勻分布在分配的CPU核心上。如果不實(shí)現(xiàn)前者，一些Operator可能會(huì)過(guò)載或過(guò)度配置，分別成為性能瓶頸或浪費(fèi)計(jì)算資源。如果不實(shí)現(xiàn)后者，一些CPU內(nèi)核將會(huì)過(guò)載，而其他CPU內(nèi)核將未得到充分利用，從而導(dǎo)致性能下降。我們將Operator縮放和負(fù)載平衡的機(jī)制稱(chēng)為彈性。為了在動(dòng)態(tài)工作負(fù)載下保持高性能，快速?gòu)椥允且豁?xiàng)至關(guān)重要的要求。

2.2 The Executor-Centric Paradigm

?????? 表1總結(jié)了現(xiàn)有兩種彈性范式的主要特征：靜態(tài)和以資源為中心的方法。 靜態(tài)方法使用固定數(shù)量的執(zhí)行程序?qū)崿F(xiàn)每個(gè)Operator，并使用靜態(tài)Operator Key Space在執(zhí)行程序之間分配工作負(fù)載。 每個(gè)執(zhí)行程序由綁定到指定CPU核心的單個(gè)數(shù)據(jù)處理線程組成。 由于靜態(tài)Key分區(qū)和CPU內(nèi)核與執(zhí)行器的一對(duì)一綁定，靜態(tài)方法簡(jiǎn)化了系統(tǒng)實(shí)現(xiàn)，并在大多數(shù)最先進(jìn)的系統(tǒng)中采用[30,43]。 但是，由于既不能平衡分配的CPU內(nèi)核的工作負(fù)載，也不能調(diào)整分配給特定Operator的CPU內(nèi)核數(shù)量，因此這種方法對(duì)分區(qū)模式非常敏感，并且由于缺乏彈性而在動(dòng)態(tài)工作負(fù)載下效率低下。

?????? 以資源為中心的方法通過(guò)支持動(dòng)態(tài)Operator級(jí)Key分區(qū)來(lái)解決靜態(tài)方法的限制，同時(shí)遵循與靜態(tài)方法相同的執(zhí)行程序?qū)崿F(xiàn)。 憑借Operator級(jí)Key重新分區(qū)的功能，以資源為中心的方法實(shí)現(xiàn)了彈性，因?yàn)樗梢詫⒁恍㎏ey及其相應(yīng)的工作負(fù)載從重載執(zhí)行程序遷移到負(fù)載較輕的執(zhí)行程序，以平衡工作負(fù)載，或從現(xiàn)有執(zhí)行程序遷移到新創(chuàng)建的執(zhí)行者以擴(kuò)展Operator。 但是，如引言部分所述，此Operator級(jí)Key重新分區(qū)是一個(gè)耗時(shí)的過(guò)程，在此過(guò)程中需要昂貴的全局同步來(lái)遷移狀態(tài)并更新所有上游執(zhí)行程序的路由表。 因此，以資源為中心的方法不能實(shí)現(xiàn)快速?gòu)椥?#xff0c;只能解決非常有限的工作量動(dòng)態(tài)。

?????? 為了實(shí)現(xiàn)快速?gòu)椥?#xff0c;我們提出了一種新的執(zhí)行范式：以executor-centric的方法。我們的想法來(lái)自觀察到Operator級(jí)別的Key重新分區(qū)太昂貴而無(wú)法實(shí)現(xiàn)快速?gòu)椥浴?strong>因此，以執(zhí)行器為中心的方法使用靜態(tài)Operator級(jí)別的Key分區(qū)，但將每個(gè)執(zhí)行程序?qū)崿F(xiàn)為彈性的構(gòu)建塊以處理工作負(fù)載波動(dòng)。特別是，每個(gè)執(zhí)行程序都旨在通過(guò)動(dòng)態(tài)創(chuàng)建或刪除數(shù)據(jù)處理線程來(lái)利用各種計(jì)算資源。因此，為了實(shí)現(xiàn)負(fù)載平衡和Operator擴(kuò)展，系統(tǒng)可以為每個(gè)彈性執(zhí)行器動(dòng)態(tài)分配適當(dāng)數(shù)量的CPU內(nèi)核，而不是執(zhí)行昂貴的Operator Key重新分區(qū)。與Operator Key重新分區(qū)相比，可以有效地實(shí)現(xiàn)CPU內(nèi)核的重新分配和內(nèi)部執(zhí)行器負(fù)載平衡，因?yàn)樗鼈儾恍枰魏蜲perator間或執(zhí)行器間同步。有趣的是，我們的新方法通過(guò)避免全局同步實(shí)現(xiàn)了快速?gòu)椥浴?/span>

2.3 Overview of Elasticutor Framework

?????? 遵循以執(zhí)行者為中心的方法，我們?cè)O(shè)計(jì)了專(zhuān)注于支持有狀態(tài)流處理的Elasticutor。 為了處理流系統(tǒng)上的大規(guī)模數(shù)據(jù)，我們假設(shè)數(shù)據(jù)和狀態(tài)是在Key Space下定義的，基于哪些分區(qū)數(shù)據(jù)流和狀態(tài)可以由分布式計(jì)算單元并行處理和維護(hù)。 我們假設(shè)Key Space足夠細(xì)粒度，以便甚至可以在越來(lái)越多的計(jì)算資源（即CPU核心）上分配和平衡扭曲的工作負(fù)載。 對(duì)于像Heron，Flink和Samza這樣的其他最先進(jìn)的流處理系統(tǒng)，還需要這種假設(shè)來(lái)實(shí)現(xiàn)高度并行化的有狀態(tài)流處理。

?????? 我們的設(shè)計(jì)目標(biāo)是實(shí)現(xiàn)實(shí)時(shí)響應(yīng)，歸結(jié)為保證低延遲。 然而，過(guò)度延遲可能是由于較高的數(shù)據(jù)到達(dá)率導(dǎo)致的系統(tǒng)資源不足，或?qū)е鹿ぷ髫?fù)載不平衡的低效資源分配和調(diào)度。 前者需要資源縮放，而后者則不需要。 基于關(guān)注點(diǎn)分離原則，我們將Elasticutor設(shè)計(jì)為兩級(jí)架構(gòu)，如圖2所示。

?????? 高級(jí)調(diào)度程序（在第4節(jié)中描述）處理可能在一段時(shí)間內(nèi)激增的動(dòng)態(tài)工作負(fù)載，在此期間現(xiàn)有系統(tǒng)容量不足并需要擴(kuò)展。不需要過(guò)度配置，但我們假設(shè)可以從基于云的平臺(tái)按需獲取資源。我們假設(shè)工作負(fù)載的總體激增不會(huì)太頻繁發(fā)生，例如，在幾分鐘到幾小時(shí)的時(shí)間范圍內(nèi)。動(dòng)態(tài)調(diào)度程序確定每個(gè)彈性執(zhí)行程序應(yīng)在瞬時(shí)工作負(fù)載下提供的所需CPU核心數(shù)。它采用基于排隊(duì)網(wǎng)絡(luò)的性能模型，并使用彈性執(zhí)行器的收集性能Metrics指標(biāo)作為輸入來(lái)生成資源分配決策。根據(jù)現(xiàn)有的核心到執(zhí)行器分配和集群中CPU核心的可用性，調(diào)度程序改進(jìn)了分配以適應(yīng)新的資源分配計(jì)劃，同時(shí)考慮了CPU重新分配開(kāi)銷(xiāo)和計(jì)算資源的位置。

?????? 每個(gè)low-level執(zhí)行程序（在第3節(jié)中描述）被設(shè)計(jì)為一個(gè)輕量級(jí)，自包含的分布式子系統(tǒng)，稱(chēng)為彈性執(zhí)行程序，負(fù)責(zé)處理固定key subspace下的輸入。 為適應(yīng)工作負(fù)載波動(dòng)，彈性執(zhí)行器可以使用動(dòng)態(tài)數(shù)量的CPU核心，可能來(lái)自多個(gè)節(jié)點(diǎn)，由動(dòng)態(tài)調(diào)度程序決定。 為了在工作負(fù)載波動(dòng)的情況下充分利用其分配的CPU內(nèi)核，彈性執(zhí)行器具有高效的內(nèi)部負(fù)載平衡機(jī)制，可以在更短的時(shí)間范圍內(nèi)將分配的CPU內(nèi)核的輸入流的計(jì)算均勻分布。

?????? 以有狀態(tài)處理為目標(biāo)的流處理系統(tǒng)的設(shè)計(jì)空間還包括諸如狀態(tài)大小和數(shù)據(jù)流特征之類(lèi)的維度，即每元組的計(jì)算和大小，以及key space下數(shù)據(jù)流的傾斜程度和動(dòng)態(tài)性。我們將討論權(quán)衡Elasticutor與第5節(jié)中的變更方法進(jìn)行比較。

3. ELASTIC EXECUTOR

?????? 為了有效地利用CPU資源，彈性執(zhí)行器被設(shè)計(jì)為適應(yīng)兩種動(dòng)態(tài)：1）key分配的變化和2）CPU core重新分配，如圖3所示。前者來(lái)自輸入流的波動(dòng)，而后者由調(diào)度程序確定全局優(yōu)化。 為了在其計(jì)算資源上分配工作負(fù)載，彈性執(zhí)行程序?yàn)槊總€(gè)分配的CPU核心創(chuàng)建任務(wù)，并在其上分配輸入數(shù)據(jù)元組。 在CPU重新分配時(shí)，將創(chuàng)建新任務(wù)或刪除現(xiàn)有任務(wù)。 這兩種動(dòng)態(tài)都會(huì)在任務(wù)之間引入不平衡的工作負(fù)載，從而導(dǎo)致資源利用不足或性能下降。 因此，一個(gè)中心設(shè)計(jì)問(wèn)題是如何在存在這種動(dòng)態(tài)的情況下在任務(wù)之間保持平衡的工作負(fù)載分配。

3.1 Components and Working Mechanism

?????? 如圖4所示，彈性執(zhí)行器被實(shí)現(xiàn)為輕量級(jí)，自包含的分布式子系統(tǒng)，其可以利用多個(gè)物理節(jié)點(diǎn)上的計(jì)算資源。 每個(gè)彈性執(zhí)行程序主要駐留在一個(gè)稱(chēng)為其本地節(jié)點(diǎn)的物理節(jié)點(diǎn)中，在該節(jié)點(diǎn)中，它運(yùn)行本地主進(jìn)程以接收輸入元組并發(fā)送輸出元組。 對(duì)于每個(gè)分配的CPU核心，在該過(guò)程中創(chuàng)建作為數(shù)據(jù)處理線程實(shí)現(xiàn)的任務(wù)。 要在遠(yuǎn)程節(jié)點(diǎn)上使用CPU核心，可以創(chuàng)建遠(yuǎn)程進(jìn)程來(lái)托管遠(yuǎn)程數(shù)據(jù)處理的遠(yuǎn)程任務(wù)。

Intra-Executor Routing: 我們采用雙層設(shè)計(jì)，在圖4中央結(jié)構(gòu)中顯示的路由表中實(shí)現(xiàn)，根據(jù)瞬時(shí)工作負(fù)載分配動(dòng)態(tài)地將輸入元組映射到任務(wù)。第一層使用靜態(tài)散列函數(shù)將key Sub Space靜態(tài)劃分為Shard;第二層顯式維護(hù)動(dòng)態(tài)Shard到任務(wù)映射，該映射在Shard重新分配時(shí)更新。我們?cè)诖至６榷敲總€(gè)Key的基礎(chǔ)上平衡工作負(fù)載，主要是因?yàn)榧?xì)粒度方法需要維護(hù)每個(gè)Key的工作負(fù)載，因此遭受高內(nèi)存消耗。Shard數(shù)量的選擇提供了負(fù)載平衡質(zhì)量和維護(hù)開(kāi)銷(xiāo)之間的權(quán)衡。然而，在實(shí)踐中，合理數(shù)量的Shard（例如，任務(wù)數(shù)量的4或8倍）實(shí)現(xiàn)了良好的平衡質(zhì)量，同時(shí)保持低維護(hù)開(kāi)銷(xiāo)。我們將在5.3節(jié)討論Shard數(shù)如何影響某些極端設(shè)置中的系統(tǒng)性能。

Executor-Level Fault Tolerance：在流處理系統(tǒng)中已經(jīng)廣泛研究了容錯(cuò)[8,11,14,35,49]，因此既不是本文的重點(diǎn)也不是本文的貢獻(xiàn)。 在這里，我們只討論如何從故障中恢復(fù)每個(gè)彈性執(zhí)行器的遠(yuǎn)程任務(wù)，以便Elasticutor可以利用最先進(jìn)的狀態(tài)檢查點(diǎn)技術(shù)，如流水線快照協(xié)議[11]，實(shí)現(xiàn)容錯(cuò)。

?????? 彈性執(zhí)行程序的主要過(guò)程在邏輯上維護(hù)其任務(wù)狀態(tài)的主副本。默認(rèn)情況下，狀態(tài)在主進(jìn)程的內(nèi)存中維護(hù)以進(jìn)行高效訪問(wèn)，但是當(dāng)狀態(tài)太大而無(wú)法容納在內(nèi)存中時(shí)，也可以將狀態(tài)存儲(chǔ)在外部存儲(chǔ)中。主進(jìn)程還用增加的元組ID標(biāo)記每個(gè)輸入數(shù)據(jù)元組。對(duì)于每個(gè)遠(yuǎn)程任務(wù)T，彈性執(zhí)行程序維護(hù)一個(gè)待處理的元組隊(duì)列以備份發(fā)送到T的數(shù)據(jù)元組，并且值ts表示已經(jīng)處理了ID小于ts的所有數(shù)據(jù)元組，并且處理這些元組產(chǎn)生的狀態(tài)更新已被沖回主副本。每個(gè)遠(yuǎn)程任務(wù)周期性地，例如每10秒，將其本地狀態(tài)的更新與tmax（即，已經(jīng)處理的元組的最大ID）一起發(fā)送到彈性執(zhí)行器的主進(jìn)程。在從遠(yuǎn)程任務(wù)T接收到狀態(tài)更新時(shí)，彈性執(zhí)行器更新?tīng)顟B(tài)的主副本，在T的待處理隊(duì)列中移除ID不大于tmax的元組，并更新ts = tmax。當(dāng)遠(yuǎn)程任務(wù)T失敗時(shí)，彈性執(zhí)行程序使用狀態(tài)的主副本創(chuàng)建新任務(wù)，并通過(guò)在待處理隊(duì)列T中重放大于ts的ID的元組來(lái)開(kāi)始執(zhí)行新任務(wù)。

3.2 Consistent Workload Redistribution

?????? 雖然狀態(tài)共享提高了Shard重新分配的效率，但需要注意保證一致性。 一般而言，盡管使用與以資源為中心的方法的Key重新分區(qū)類(lèi)似的過(guò)程，我們通過(guò)利用以執(zhí)行器為中心的方法啟用的Operator間和執(zhí)行者間的獨(dú)立性來(lái)實(shí)現(xiàn)具有狀態(tài)一致性的有效分片重新分配。

?????? 考慮圖4中的情況，其中元組t1處于任務(wù)T2的待處理隊(duì)列中，元組t2剛剛到達(dá)執(zhí)行器的主進(jìn)程，并且元組t3將由上游執(zhí)行器發(fā)出。 假設(shè)所有三個(gè)元組都屬于shard r4。 如果在處理t1之前或在更新t2和t3的路由之前將shard r4從源任務(wù)T2重新分配給新的目標(biāo)任務(wù)，則狀態(tài)將變得不一致。 特別地，如果目的地任務(wù)是本地的，例如T1，那么t2可能在t1之前被處理，違反了順序處理的要求。 如果目的地任務(wù)是遠(yuǎn)程的，例如T0，則由t1對(duì)狀態(tài)的修改將丟失。

Inter-Operator Consistent Routing：為了保證從上游Operator到分配的任務(wù)所在的正確進(jìn)程的一致路由（例如t3），彈性執(zhí)行器在其本地主進(jìn)程中實(shí)現(xiàn)接收器守護(hù)進(jìn)程，作為來(lái)自上游Operator的所有元組的單一入口。接收器根據(jù)內(nèi)部路由表將元組路由到適當(dāng)?shù)谋镜鼗蜻h(yuǎn)程任務(wù)。類(lèi)似地，發(fā)射器守護(hù)程序在主進(jìn)程中實(shí)現(xiàn)為執(zhí)行程序的單個(gè)退出，以將任務(wù)生成的輸出元組轉(zhuǎn)發(fā)給下游Operator。遠(yuǎn)程過(guò)程僅與彈性執(zhí)行器的主過(guò)程上的接收器和發(fā)射器通信。因此，無(wú)論在彈性執(zhí)行器中的任務(wù)之間如何動(dòng)態(tài)重新分配分片，上游和下游Operator總是通過(guò)其接收器和發(fā)送器向執(zhí)行器發(fā)送元組或從接收器接收元組，從而避免由分片重新分配引起的任何Operator間同步。相反，以資源為中心的方法通過(guò)Operator級(jí)key space重新分區(qū)來(lái)重新分配工作負(fù)載，從而導(dǎo)致與所有上游執(zhí)行程序同步。

?????? 請(qǐng)注意，與以資源為中心的方法相比，來(lái)自上游執(zhí)行器的元組直接路由到下游Operator，Elasticutor可能涉及接收器/發(fā)射器和遠(yuǎn)程任務(wù)之間的額外遠(yuǎn)程數(shù)據(jù)傳輸。 這是我們?yōu)閷?shí)現(xiàn)快速?gòu)椥远龀龅臋?quán)衡。 在典型的工作負(fù)載中，遠(yuǎn)程數(shù)據(jù)傳輸不是性能瓶頸，如圖13所示。在5.3節(jié)中，我們通過(guò)正確配置Operator的執(zhí)行器數(shù)量，討論如何避免/減少某些極端工作負(fù)載中的遠(yuǎn)程數(shù)據(jù)傳輸。

Intra-Executor State Consistency: 為了保證在重新分區(qū)Shard期間的狀態(tài)一致性，彈性檢測(cè)器采用類(lèi)似于以資源為中心的方法中使用的Operator級(jí)重新分區(qū)的Key重新分區(qū)過(guò)程，但不涉及任何全局同步。關(guān)鍵是要確保a）在將shard狀態(tài)遷移到目標(biāo)任務(wù)之前必須處理掛起的元組，即SOurce任務(wù)中排隊(duì)的分片的未處理元組。 b）具有相同Key的元組不會(huì)同時(shí)在任何兩個(gè)任務(wù)中處理。在圖4中Shard r4的重新分配期間，暫停了元組r4的路由，并將標(biāo)簽元組發(fā)送到其Source taskT2。由于任務(wù)以先來(lái)先服務(wù)的方式處理其輸入元組，因此當(dāng)T2從其待處理隊(duì)列中拉出標(biāo)簽元組時(shí)，保證已經(jīng)發(fā)送到T2的任何待處理元組唄處理。之后，r4的狀態(tài)將遷移到目標(biāo)任務(wù)。如果將分片重新分配給其源任務(wù)的本地任務(wù)，則省略狀態(tài)遷移。在狀態(tài)遷移之后，在恢復(fù)r4的元組的路由之前，在路由表中更新Shard到任務(wù)映射。

Discussions：值得注意的是，我們提出的以執(zhí)行為中心的范例適用于其他現(xiàn)有的分布式流系統(tǒng)，例如Apache Flink，Apache Heron和Apache Samza，其中有狀態(tài)處理可以通過(guò)在key space下劃分狀態(tài)和數(shù)據(jù)來(lái)并行化。 對(duì)于無(wú)狀態(tài)應(yīng)用程序，我們的方法仍然可以應(yīng)用，但可能不一定是最佳選擇，因?yàn)橥ㄟ^(guò)簡(jiǎn)單地以循環(huán)方式發(fā)送元組或者向負(fù)載最少的執(zhí)行程序發(fā)送元組可以輕松實(shí)現(xiàn)負(fù)載平衡。

雖然我們的方法不適用于基于批處理的系統(tǒng)，但我們的雙層負(fù)載平衡設(shè)計(jì)與小批量定向的Spark Streaming [50]采用的方法有一些相似之處。 兩個(gè)主要的區(qū)別是：1）我們?cè)O(shè)計(jì)一個(gè)額外的中間層碎片提供了維護(hù)成本和平衡負(fù)載之間的權(quán)衡，2）我們的測(cè)量和平衡設(shè)計(jì)對(duì)流量系統(tǒng)更自然，其中操作過(guò)程輸入元組 而不是基于小批量。

4. DYNAMIC SCHEDULER

?????? 動(dòng)態(tài)調(diào)度程序的目標(biāo)是通過(guò)在不斷變化的工作負(fù)載下自適應(yīng)地將CPU核心分配給彈性執(zhí)行程序來(lái)滿(mǎn)足用戶(hù)定義的延遲要求。 通過(guò)使用由系統(tǒng)測(cè)量的瞬時(shí)性能指標(biāo)，調(diào)度程序首先根據(jù)排隊(duì)網(wǎng)絡(luò)模型估計(jì)每個(gè)執(zhí)行程序所需的核心數(shù)量，并進(jìn)一步（重新）將物理核心分配給執(zhí)行者，以便最小化重新分配開(kāi)銷(xiāo) 并最大化執(zhí)行程序內(nèi)的計(jì)算位置。

4.1 Model-Based Resource Allocation

?????? 我們將m個(gè)彈性執(zhí)行器的拓?fù)銭 = {1，··，m}建模為Jackson網(wǎng)絡(luò)，其中每個(gè)執(zhí)行器j∈E被視為M / M / kj系統(tǒng)[42]，其中kj表示分配給j的CPU核心數(shù)。 輸入流的平均處理等待時(shí)間，表示為E [T]，可以作為資源分配決策k的函數(shù)來(lái)計(jì)算。

?????? 其中λ0表示輸入流的到達(dá)率，Tj和λj分別表示執(zhí)行器j的平均處理時(shí)間和到達(dá)率。 當(dāng)kj>λj/μj時(shí)，每個(gè)E [Tj]（kj）是有界的，其中μj表示彈性執(zhí)行器j的處理速率，并且可以作為由系統(tǒng)測(cè)量的參數(shù)λ0，{λj}和{μj}的函數(shù)來(lái)計(jì)算。基于等式（1），調(diào)度器嘗試找到分配k以確保E [T]不大于用戶(hù)指定的等待時(shí)間目標(biāo)Tmax，同時(shí)最小化CPU核心的總數(shù)。 特別地，每個(gè)kj被初始化為λj/μj+ 1，這是使系統(tǒng)穩(wěn)定的最低要求。 我們重復(fù)向向量k中的值加1，導(dǎo)致E [T]的最顯著減少，直到E [T]≤Tmax或íkj超過(guò)可用CPU資源的數(shù)量。 這個(gè)貪心算法在找到解k時(shí)已經(jīng)證明是最優(yōu)的[22]。

4.2 CPU-to-Executor Assignment

?????? 性能模型僅建議新的分配，即每個(gè)執(zhí)行者需要的CPU核心數(shù)量，這是由工作負(fù)載波動(dòng)引起的; 調(diào)度程序仍然需要通過(guò)更新現(xiàn)有的核心到執(zhí)行程序分配來(lái)適應(yīng)新的分配計(jì)劃。 新分配計(jì)劃的CPU重新分配是系統(tǒng)性能的關(guān)鍵，因?yàn)樗赡軙?huì)引入1）轉(zhuǎn)換期間的狀態(tài)遷移成本，以及2）之后的遠(yuǎn)程數(shù)據(jù)傳輸成本。 例如，在重新分配CPU內(nèi)核時(shí)，彈性檢測(cè)器會(huì)創(chuàng)建一個(gè)新任務(wù)，如果CPU內(nèi)核遠(yuǎn)離彈性執(zhí)行程序，則涉及狀態(tài)遷移和將來(lái)的遠(yuǎn)程數(shù)據(jù)傳輸。 為了優(yōu)化執(zhí)行效率，我們搜索最小化遷移成本的CPU到執(zhí)行器分配，同時(shí)限制計(jì)算局部性以限制未來(lái)的遠(yuǎn)程數(shù)據(jù)傳輸成本。

?????? 為了模擬遷移成本，我們考慮一個(gè)n個(gè)節(jié)點(diǎn)的集群，其中每個(gè)節(jié)點(diǎn)都有ci CPU核心。對(duì)于任何執(zhí)行器j∈E，我們用I（j）表示其主進(jìn)程所在的節(jié)點(diǎn)，并且通過(guò)列向量xj =（x1j，...，xnj）T表示在所有節(jié)點(diǎn)上分配給它的核的數(shù)量。我們將jj = i = 1 xi jas定義為j的指定核心總數(shù)，并用矩陣X =（x1，···，xm）表示CPU到執(zhí)行器的分配。給定任何新的分配k，從現(xiàn)有分配X到新分配X的轉(zhuǎn)換需要執(zhí)行一組CPU分配/解除分配。核心重新分配的開(kāi)銷(xiāo)由狀態(tài)遷移成本決定，狀態(tài)遷移成本與跨網(wǎng)絡(luò)移動(dòng)的狀態(tài)大小成比例。我們用sj表示任何執(zhí)行者j的聚合狀態(tài)大小。為簡(jiǎn)單起見(jiàn)，我們假設(shè)彈性執(zhí)行器的分片均勻分布在分配的CPU核心上;因此，與每個(gè)CPU核心相關(guān)的狀態(tài)數(shù)據(jù)量大約是sj / Xj。給定任何分配k，可用核心c和現(xiàn)有分配X~，我們按如下方式制定CPU分配問(wèn)題。

上述優(yōu)化問(wèn)題最小化了從現(xiàn)有賦值X到新賦值X的遷移成本C（X | X?），其中求和中的每個(gè)項(xiàng)都測(cè)量執(zhí)行者j將其狀態(tài)遷移出節(jié)點(diǎn)i的成本。約束包括（a）CPU核心的數(shù)量，（b）分配要求和（c）計(jì)算局部性，即，要求分配給執(zhí)行器的集合E（φ）的所有核心都在其本地節(jié)點(diǎn)上。系統(tǒng)通過(guò)其總輸入和輸出數(shù)據(jù)速率除以核心數(shù)kj來(lái)測(cè)量任何執(zhí)行器j的瞬時(shí)每核數(shù)據(jù)強(qiáng)度，并且E（φ）表示數(shù)據(jù)強(qiáng)度高于閾值的執(zhí)行器集合。 。因?yàn)槿绻峙涞暮诵氖沁h(yuǎn)程的，數(shù)據(jù)密集型執(zhí)行程序?qū)a(chǎn)生更高的網(wǎng)絡(luò)成本，我們通過(guò)避免將遠(yuǎn)程核心分配給E（φ）的成員來(lái)強(qiáng)制執(zhí)行計(jì)算局部性。這個(gè)整數(shù)規(guī)劃問(wèn)題可以簡(jiǎn)化為NP-hard多處理器調(diào)度問(wèn)題[23]。因此，我們?cè)O(shè)計(jì)了一個(gè)有效的貪婪算法1來(lái)找到近似解。對(duì)于任何賦值X，我們將E + = {j∈E| Xj <kj}定義為欠配置執(zhí)行器的集合，E +Δ= {j∈E+∩E（φ）}作為數(shù)據(jù)密集型執(zhí)行器的子集，并且E- = {j∈E| Xj> kj}是一組過(guò)度供應(yīng)的執(zhí)行者。我們使用C + i j（X）和

C-ij（X）分別表示在節(jié)點(diǎn)i上分配/取消分配CPU核心到執(zhí)行器j的開(kāi)銷(xiāo)，可以導(dǎo)出為C + ij（X）= sj（Xj-xi j）/（Xj（ Xj + 1））和C-ij（X）= sj（Xj-xi j）/（Xj（Xj-1））。

算法1按數(shù)據(jù)強(qiáng)度按降序?qū) +中的執(zhí)行程序進(jìn)行排序，并嘗試通過(guò)從其他執(zhí)行程序解除分配核心，逐個(gè)為每個(gè)執(zhí)行程序j分配目標(biāo)CPU核心數(shù)。具體來(lái)說(shuō)，如果彈性執(zhí)行器j是數(shù)據(jù)密集型的，即j∈E（φ），它只接受節(jié)點(diǎn)i = I（j）上的CPU核心，以避免創(chuàng)建遠(yuǎn)程任務(wù)。因此，在所有非數(shù)據(jù)密集型執(zhí)行程序中，算法在節(jié)點(diǎn)I（j）上找到一個(gè)CPU核心，可以用最小的釋放開(kāi)銷(xiāo)重新分配給j（第7行）。相反，如果j不是數(shù)據(jù)密集型，則它接受任何節(jié)點(diǎn)上的CPU核心。該算法搜索E-中的所有執(zhí)行程序，以獲得具有CPU內(nèi)核的執(zhí)行程序，該內(nèi)核可以通過(guò)最小的釋放和分配開(kāi)銷(xiāo)重新分配給j（第9行）。在任何一種情況下，如果找到這樣的有效核心重新分配，則算法將其添加到新的賦值X中;否則，它返回FAIL，這表示沒(méi)有找到可行的解決方案，并且暗示需要更高的數(shù)據(jù)不敏感閾值φ來(lái)獲得可行的解決方案。

?????? 算法1按數(shù)據(jù)強(qiáng)度按降序?qū) +中的執(zhí)行程序進(jìn)行排序，并嘗試通過(guò)從其他執(zhí)行程序解除分配核心，逐個(gè)為每個(gè)執(zhí)行程序j分配目標(biāo)CPU核心數(shù)。具體來(lái)說(shuō)，如果彈性執(zhí)行器j是數(shù)據(jù)密集型的，即j∈E（φ），它只接受節(jié)點(diǎn)i = I（j）上的CPU核心，以避免創(chuàng)建遠(yuǎn)程任務(wù)。因此，在所有非數(shù)據(jù)密集型執(zhí)行程序中，算法在節(jié)點(diǎn)I（j）上找到一個(gè)CPU核心，可以用最小的釋放開(kāi)銷(xiāo)重新分配給j（第7行）。相反，如果j不是數(shù)據(jù)密集型，則它接受任何節(jié)點(diǎn)上的CPU核心。該算法搜索E-中的所有執(zhí)行程序，以獲得具有CPU內(nèi)核的執(zhí)行程序，該內(nèi)核可以通過(guò)最小的釋放和分配開(kāi)銷(xiāo)重新分配給j（第9行）。在任何一種情況下，如果找到這樣的有效核心重新分配，則算法將其添加到新的賦值X中;否則，它返回FAIL，這表示沒(méi)有找到可行的解決方案，并且暗示需要更高的數(shù)據(jù)不敏感閾值φ來(lái)獲得可行的解決方案。

?????? φ的選擇提供了公式4.2的可行性與彈性執(zhí)行器的計(jì)算局部性之間的權(quán)衡。由于動(dòng)態(tài)分配算法非常有效，我們使用低默認(rèn)值φ=?φ運(yùn)行算法。如果沒(méi)有找到可行的解決方案，我們將φ加倍并重新運(yùn)行算法，直到我們找到一個(gè)。在我們的實(shí)驗(yàn)中，我們將φ~設(shè)置為512 KB / s，低于該值時(shí)，計(jì)算局部性的好處可以忽略不計(jì)。

Discussions：我們的動(dòng)態(tài)調(diào)度程序設(shè)計(jì)適用于使用連續(xù)運(yùn)算符的流處理，并遵循數(shù)據(jù)流模型[5]。調(diào)度程序確定每個(gè)執(zhí)行程序滿(mǎn)足延遲要求所需的資源，并計(jì)算資源分配以最大限度地降低狀態(tài)遷移成本。在這個(gè)級(jí)別工作的其他調(diào)度程序包括Flink的DS2 [28]，Heron的Dhalion [21]，Storm的RAS [36]等等。相比之下，基于云的資源管理系統(tǒng)（如YARN [44]和Mesos [25]）更加以集群為中心[26,27]，即它們主要旨在管理不同應(yīng)用程序之間的集群資源。他們通常會(huì)收到應(yīng)用程序經(jīng)理的資源需求，并根據(jù)效率和公平等標(biāo)準(zhǔn)來(lái)決定如何配置資源。通常開(kāi)發(fā)協(xié)商器/協(xié)調(diào)器模塊用于協(xié)助不同級(jí)別的調(diào)度器之間的交互。典型的例子包括Storm-on-Yarn [3]和Flink-on-Yarn [2]。

5. PERFORMANCE EVALUATION

?????? 我們?cè)贏pache Storm上大約10,000行Java中實(shí)現(xiàn)了Elasticutor的原型[43]。 Elasticutor的源代碼可在[4]獲得。 Storm是一種流行的分布式流處理系統(tǒng)，它暴露了低級(jí)API，例如Bolt API。 這對(duì)于原型研究思想來(lái)說(shuō)相對(duì)容易一些。 Storm遵循靜態(tài)方法，其操作符由用戶(hù)通過(guò)抽象類(lèi)Bolt實(shí)現(xiàn)。 我們添加了一個(gè)新的抽象類(lèi)ElasticBolt，它提供了與Bolt相同的編程接口，但是向用戶(hù)空間公開(kāi)了一個(gè)新的狀態(tài)訪問(wèn)接口。 對(duì)于任何定義為ElasticBolt的運(yùn)算符，Elasticutor創(chuàng)建了許多具有內(nèi)置狀態(tài)管理，度量標(biāo)準(zhǔn)Metrics測(cè)量和彈性功能的彈性執(zhí)行器。 動(dòng)態(tài)調(diào)度程序?qū)崿F(xiàn)為在Storm的主節(jié)點(diǎn)（nimbus）上運(yùn)行的守護(hù)程序進(jìn)程。

?????? 我們的實(shí)驗(yàn)在Amazon EC2上進(jìn)行，具有32個(gè)t2.2x大型實(shí)例（節(jié)點(diǎn)），每個(gè)實(shí)例具有8個(gè)CPU核心和32GB RAM，運(yùn)行Ubuntu 16.04。網(wǎng)絡(luò)是1Gbps以太網(wǎng)。在所有方法下，執(zhí)行器以循環(huán)方式分配給節(jié)點(diǎn)。除非另有說(shuō)明，否則Elasticutor每個(gè)運(yùn)算符使用32個(gè)彈性執(zhí)行程序，每個(gè)執(zhí)行程序使用256個(gè)分片（每個(gè)運(yùn)算符8192個(gè)分片）。為了公平比較，我們?yōu)殪o態(tài)方法中的Operator創(chuàng)建了足夠的執(zhí)行程序，以充分利用集群中的所有CPU核心;并將RC方法中的key spcae分區(qū)的粒度設(shè)置為每個(gè)運(yùn)算符8192個(gè)分片，與Elasticutor中的相同。為確保系統(tǒng)穩(wěn)定性，Storm，Heron和Flink等現(xiàn)有流系統(tǒng)實(shí)施反壓機(jī)制，以控制Operator的輸入速率。為了關(guān)注系統(tǒng)性能，我們?cè)u(píng)估壓力情況，其中足夠高的到達(dá)率使輸入隊(duì)列保持非空并且可能觸發(fā)Storm的背壓機(jī)制。

?????? 我們?cè)?.3節(jié)中詳細(xì)討論了有狀態(tài)流處理設(shè)計(jì)空間中的Elasticutor。在本節(jié)中，我們將Elasticutor的性能與靜態(tài)方法（默認(rèn)Storm）和以資源為中心（RC）方法的性能進(jìn)行比較。第2.2節(jié)總結(jié)了這三種方法的主要區(qū)別。我們通過(guò)啟用創(chuàng)建/刪除執(zhí)行程序和Operator級(jí)Key重新分區(qū)來(lái)實(shí)現(xiàn)基于Storm的RC。為了公平比較，RC使用與Elasticutor相同的性能模型，負(fù)載平衡算法和進(jìn)程內(nèi)狀態(tài)共享機(jī)制。我們將評(píng)估Elasticutor在設(shè)計(jì)空間中的不同維度所做出的性能和權(quán)衡，包括狀態(tài)大小，每元組計(jì)算和大小，以及數(shù)據(jù)流的偏度和動(dòng)態(tài)性。通常，只要有足夠的計(jì)算資源可用于系統(tǒng)中的擴(kuò)展，Elasticutor amis的設(shè)計(jì)就可以容納計(jì)算密集型工作負(fù)載。但是，由于遠(yuǎn)程任務(wù)的引入可能會(huì)導(dǎo)致數(shù)據(jù)傳輸和狀態(tài)遷移開(kāi)銷(xiāo)和延遲，我們的設(shè)計(jì)假定工作負(fù)載在元組大小和狀態(tài)大小方面不會(huì)過(guò)于數(shù)據(jù)密集，并且網(wǎng)絡(luò)帶寬容量確實(shí)如此不成為瓶頸。我們假設(shè)在關(guān)鍵領(lǐng)域的數(shù)據(jù)分布中表現(xiàn)出的傾斜度是一種規(guī)范，我們關(guān)注的是更具挑戰(zhàn)性的情況，其中數(shù)據(jù)傾斜度也會(huì)突然變化，這可以在股票交易數(shù)據(jù)集和評(píng)估中顯示出來(lái)。

5.1 Micro-Benchmarking

?????? 在本小節(jié)中，我們使用一個(gè)簡(jiǎn)單而有代表性的拓?fù)浣Y(jié)構(gòu)，如圖5所示，它允許輕松控制工作負(fù)載特性，例如輸入速率，計(jì)算成本和數(shù)據(jù)分布。拓?fù)溆缮善骱陀?jì)算器組成，輸入數(shù)據(jù)流由生成器饋送到計(jì)算器進(jìn)行處理。我們確保數(shù)據(jù)生成速率使計(jì)算器的輸入隊(duì)列飽和。計(jì)算器運(yùn)算符中每個(gè)元組的處理時(shí)間遵循正態(tài)分布N（μ，δ2=0.5μ）。通過(guò)在執(zhí)行時(shí)間內(nèi)循環(huán)運(yùn)行數(shù)據(jù)加密來(lái)實(shí)現(xiàn)計(jì)算，以耗盡CPU周期并模擬計(jì)算密集型工作負(fù)載。除非另有說(shuō)明，否則每個(gè)元組由一個(gè)整數(shù)鍵和一個(gè)128字節(jié)的有效負(fù)載組成，并且處理的平均CPU成本為1 ms。密鑰空間包含10K個(gè)不同的值，其頻率遵循zipf分布[37]，偏差因子為0.5。默認(rèn)狀態(tài)大小為256MB，每個(gè)分片為32KB。為了模擬工作負(fù)載動(dòng)態(tài)，我們通過(guò)應(yīng)用每分鐘ω次隨機(jī)排列來(lái)改變?cè)MKey的頻率。

工作負(fù)載動(dòng)態(tài)的穩(wěn)健性：圖6描繪了隨著ω沿x軸變化的三種方法下的吞吐量和平均處理延遲。我們觀察到，當(dāng)工作負(fù)載是動(dòng)態(tài)的時(shí)，Elasicutor在兩個(gè)指標(biāo)方面始終優(yōu)于其他方法，即ω > 0。特別是，由于密鑰分配偏差導(dǎo)致工作負(fù)載不平衡，靜態(tài)方法的性能較差，但由于沒(méi)有執(zhí)行彈性操作，因此在所有情況下都相對(duì)穩(wěn)定。 由于RC和Elasticutor都能夠適應(yīng)偏斜的鍵分布，因此當(dāng)ω很小時(shí)，它們會(huì)大大優(yōu)于靜態(tài)。 然而，隨著ω的增加，盡管由于彈性操作成本較高而導(dǎo)致RC和Elasticutor的性能下降，但Elasticutor的性能下降是微不足道的，而RC的性能下降變大了2-3個(gè)數(shù)量級(jí)，使RC無(wú)用為ω 達(dá)到16。

?????? 為了更好地解釋?duì)刈兓瘯r(shí)三種方法的性能，我們關(guān)注ω= 2的情景，即每30秒進(jìn)行一次混洗，并繪制在圖7中1秒的滑動(dòng)時(shí)間窗口內(nèi)測(cè)量的瞬時(shí)吞吐量。我們觀察到 靜態(tài)方法的吞吐量始終遠(yuǎn)低于RC和Elasticutor的吞吐量，盡管變化不大。 由于關(guān)鍵混洗觸發(fā)彈性操作的執(zhí)行，RC和Elasticutor每30秒就會(huì)出現(xiàn)一次瞬態(tài)吞吐量降低。 然而，RC的退化要差得多，其瞬態(tài)持續(xù)時(shí)間為10到20秒，而Elasticutor的衰減僅持續(xù)1到3秒。 這解釋了隨著工作負(fù)載變得更加動(dòng)態(tài)，兩種方法中性能差距擴(kuò)大的原因。

?????? 在不同數(shù)據(jù)強(qiáng)度下的性能：為了評(píng)估工作負(fù)載的數(shù)據(jù)強(qiáng)度如何影響三種方法的性能，我們改變?cè)M大小（表示為s）和每個(gè)元組的計(jì)算成本（表示為c），并將其性能進(jìn)行比較。結(jié)果表明，在數(shù)據(jù)強(qiáng)度較高的情況下，例如，元組大小較大或每個(gè)元組的計(jì)算成本較低，由于數(shù)據(jù)傳輸開(kāi)銷(xiāo)較高，三種方法的吞吐量會(huì)下降。例如，當(dāng)c = 0.01ms且s = 2KB時(shí)，一個(gè)CPU內(nèi)核上的全速元組處理的數(shù)據(jù)傳輸要求是2Gbps，超過(guò)了網(wǎng)絡(luò)帶寬，即1Gbps，因此導(dǎo)致性能顯著下降。所有的方法。但是，Elasticutor通常對(duì)元組大小比競(jìng)爭(zhēng)對(duì)手更敏感，特別是當(dāng)計(jì)算成本極低時(shí)，例如，每個(gè)元組c = 0.01ms，因?yàn)樗哂歇?dú)特的兩級(jí)元組路由機(jī)制，從而帶來(lái)更高的數(shù)據(jù)傳輸開(kāi)銷(xiāo)數(shù)據(jù)強(qiáng)度。

?????? 不同狀態(tài)大小下的性能：圖9比較了三種方法在吞吐量和延遲方面的性能，因?yàn)闋顟B(tài)大小沿x軸變化。請(qǐng)注意，由于每個(gè)運(yùn)算符有8192個(gè)分片，因此當(dāng)每個(gè)分片的狀態(tài)大小為32MB時(shí)，運(yùn)算符的狀態(tài)大小將為256GB，這相當(dāng)大。結(jié)果表明，隨著狀態(tài)大小的增加，RC和Elasticutor的性能下降，這是由于較大的狀態(tài)大小導(dǎo)致的狀態(tài)遷移開(kāi)銷(xiāo)增加。當(dāng)狀態(tài)大小接近32MB時(shí)，作為一種極端情況，由于執(zhí)行彈性的巨大運(yùn)營(yíng)成本，Elasticutor和RC都比靜態(tài)方法表現(xiàn)更差。我們還觀察到，在相同的州規(guī)模下，Elasicutor的表現(xiàn)優(yōu)于RC方法。這表明Elasticutor中使用的技術(shù)，如狀態(tài)共享機(jī)制和動(dòng)態(tài)調(diào)度，可以有效地減少?gòu)椥圆僮髦械臓顟B(tài)遷移開(kāi)銷(xiāo)。

對(duì)執(zhí)行工作負(fù)載分配傾斜的魯棒性：實(shí)際上，由于Key分配偏差或者由于Operator級(jí)Key分區(qū)功能的不正確配置，工作負(fù)載可能無(wú)法在執(zhí)行程序之間平均分配。為了評(píng)估三種方法對(duì)傾斜執(zhí)行器工作負(fù)載分布的穩(wěn)健性，我們使用由圖10中的偏度因子α控制的變化的密鑰分布偏度來(lái)評(píng)估它們的性能。注意，α越大，密鑰分布具有越大的偏斜。例如，當(dāng)α= 0時(shí)，密鑰遵循均勻分布，而當(dāng)α≥0.8時(shí)，大多數(shù)工作負(fù)荷落入幾個(gè)密鑰。結(jié)果表明，靜態(tài)方法受到負(fù)載不平衡的影響很大，而α<0.8時(shí)，Elasticutor和RC對(duì)執(zhí)行器負(fù)載不平衡的抵抗力更強(qiáng)。主要觀察結(jié)果是，當(dāng)α≤0.6時(shí)，Elasticutor始終優(yōu)于RC，但其性能急劇下降，并且在極度偏斜的工作負(fù)荷分布下比RC差，例如α≥0.7。這表明盡管依賴(lài)于創(chuàng)建更多遠(yuǎn)程任務(wù)來(lái)處理傾斜的執(zhí)行程序工作負(fù)載分配，但Elasticutor中的執(zhí)行程序能夠處理高達(dá)α= 0.5的工作負(fù)載不平衡，而不會(huì)在運(yùn)行遠(yuǎn)程任務(wù)時(shí)引入明顯的延遲增加和吞吐量降低。但是，當(dāng)0.6≤α≤0.8時(shí)，大多數(shù)過(guò)載執(zhí)行器無(wú)法通過(guò)有效利用更多遠(yuǎn)程任務(wù)來(lái)進(jìn)一步卸載其工作負(fù)載，這主要是由于擁塞的網(wǎng)絡(luò)帶寬，因此成為性能瓶頸，導(dǎo)致系統(tǒng)吞吐量和延遲較差。

Shard reassignment cost：因?yàn)镽C方法和Elasticutor都使用碎片重新分配來(lái)平衡工作負(fù)載，我們會(huì)比較它們的成本以更好地理解產(chǎn)生的不同延遲。 圖11顯示了每個(gè)分片的平均節(jié)點(diǎn)內(nèi)和節(jié)點(diǎn)間重新分配時(shí)間，分為同步時(shí)間和狀態(tài)遷移時(shí)間。 我們觀察到RC中的碎片重新分配時(shí)間遠(yuǎn)遠(yuǎn)長(zhǎng)于Elasticutor，這主要是由于RC方法中的同步時(shí)間極長(zhǎng)。 我們還可以看到Elasticutor在狀態(tài)遷移中花費(fèi)的時(shí)間比RC短，但與同步時(shí)間相比，狀態(tài)遷移中兩種方法之間的差異較小。

?????? 為了深入了解兩種方法之間的同步時(shí)間差異，我們改變了上游執(zhí)行器的數(shù)量，并發(fā)現(xiàn)RC比Elasticutor需要2-3個(gè)更大的時(shí)間來(lái)同步，并且它們的差異隨著更多的上游執(zhí)行器而變寬， 如圖12（a）所示。 Elasticutor遵循以執(zhí)行者為中心的標(biāo)準(zhǔn)，從而避免在分片重新分配期間與上游檢查員同步。 因此，無(wú)論上游執(zhí)行器的數(shù)量如何，其同步時(shí)間約為2 ms。 相反，在RC方法中，需要更新上游執(zhí)行器的路由表，并且需要全局同步來(lái)清除執(zhí)行器和上游執(zhí)行器之間的飛行中元組。 因此，RC中的同步時(shí)間要高得多，并且隨著上游執(zhí)行器的數(shù)量而大大增加。

圖12（b）描繪了狀態(tài)大小變化時(shí)的狀態(tài)遷移時(shí)間。 我們觀察到，由于進(jìn)程內(nèi)狀態(tài)共享機(jī)制，兩種方法中的節(jié)點(diǎn)內(nèi)狀態(tài)遷移的延遲可忽略不計(jì)。 當(dāng)狀態(tài)大小達(dá)到32 MB時(shí)，節(jié)點(diǎn)間狀態(tài)遷移的時(shí)間顯著增加，其中狀態(tài)的網(wǎng)絡(luò)數(shù)據(jù)傳輸是狀態(tài)遷移過(guò)程中的主要開(kāi)銷(xiāo)。 該圖還顯示，在給定相同狀態(tài)大小的情況下，由于執(zhí)行器為中心的范例啟用了執(zhí)行器間獨(dú)立性，因此Elasticutor遷移狀態(tài)所需的時(shí)間比RC短一些。

5.2 Scalability of a Single Elastic Executor

?????? Elasticutor的主要優(yōu)點(diǎn)是它通過(guò)分配更多CPU內(nèi)核而不是通過(guò)Operator級(jí)Key Space重新分區(qū)來(lái)處理工作負(fù)載動(dòng)態(tài)。 盡管在一個(gè)合理的設(shè)置中，Operator通常有足夠的執(zhí)行器來(lái)分?jǐn)倖蝹€(gè)執(zhí)行器上的工作負(fù)載，但由于Key分配偏差，操作員不正確，執(zhí)行人員可能負(fù)載過(guò)重而需要許多遠(yuǎn)程任務(wù)。 級(jí)別分區(qū)或不必要的執(zhí)行程序。 因此，為了Elasticutor的健壯性，彈性執(zhí)行器具有良好的可伸縮性是至關(guān)重要的，即能夠有效地?cái)U(kuò)展到許多CPU核心，并且在運(yùn)行遠(yuǎn)程任務(wù)時(shí)不會(huì)引入明顯的延遲。

?????? 為了評(píng)估彈性執(zhí)行器可以有效擴(kuò)展的范圍，我們只為計(jì)算Operaotr設(shè)置了一個(gè)彈性執(zhí)行器，但逐漸分配更多的CPU核心并測(cè)量其吞吐量和處理延遲。 由于每個(gè)節(jié)點(diǎn)有8個(gè)CPU核心，因此分配的前8個(gè)核心是本地核心，后續(xù)核心是遠(yuǎn)程核心。 在我們的評(píng)估中，我們改變了彈性的數(shù)據(jù)強(qiáng)度和運(yùn)營(yíng)成本，這是影響可擴(kuò)展性的主要因素。 前者決定了遠(yuǎn)程數(shù)據(jù)傳輸在運(yùn)行遠(yuǎn)程任務(wù)時(shí)的長(zhǎng)期成本，并且與元組大小成正比，與每個(gè)元組的計(jì)算成本成反比。 后者影響執(zhí)行彈性操作的短期運(yùn)輸開(kāi)銷(xiāo)，并與規(guī)模和工作量動(dòng)態(tài)（ω）呈正相關(guān)。

?????? 圖13描繪了執(zhí)行器在不同計(jì)算成本（左）和元組大小（右）下的可伸縮性。我們觀察到單個(gè)彈性執(zhí)行器通常可以有效地?cái)U(kuò)展到整個(gè)集群（256個(gè)CPU核心），這表明遠(yuǎn)程數(shù)據(jù)傳輸?shù)某杀究梢院雎圆挥?jì)。我們還觀察到彈性執(zhí)行器無(wú)法有效地利用超過(guò)16個(gè)具有非常大的元組大小的CPU核心，例如8KB，或者非常低的計(jì)算成本，例如每個(gè)元組0.01ms，這表明巨大的遠(yuǎn)程數(shù)據(jù)傳輸鏈接到高數(shù)據(jù)強(qiáng)度可防止執(zhí)行程序擴(kuò)展。圖14顯示了彈性執(zhí)行器向外擴(kuò)展時(shí)的99％延遲。我們可以看到，在大多數(shù)情況下，由于Netty [1]啟用了有效的網(wǎng)絡(luò)數(shù)據(jù)傳輸，處理延遲不會(huì)隨著彈性執(zhí)行器的擴(kuò)展而顯著增加。然而，在數(shù)據(jù)密集型工作負(fù)載中，例如，計(jì)算成本≤0.1ms或元組大小≥2KB，隨著分配的CPU核心數(shù)超過(guò)遠(yuǎn)程數(shù)據(jù)傳輸成為性能瓶頸的點(diǎn)，等待時(shí)間大大增加。請(qǐng)注意，由于我們?cè)谌魏我粚?duì)輸入輸出執(zhí)行器之間實(shí)現(xiàn)了反壓機(jī)制，因此延遲不會(huì)無(wú)限增長(zhǎng)。

?????? 圖15顯示了彈性執(zhí)行器在各種碎片狀態(tài)大小下的可擴(kuò)展性，ω= 2（左）和16（右）。 結(jié)果表明，彈性執(zhí)行器在所有狀態(tài)尺寸下均可有效擴(kuò)展，但是32MB。 狀態(tài)較大時(shí)，狀態(tài)遷移會(huì)成為性能瓶頸，從而阻止執(zhí)行程序有效地使用遠(yuǎn)程CPU核心。 通過(guò)比較兩個(gè)子圖，我們觀察到當(dāng)ω增加到16時(shí)，由于與更高工作負(fù)載動(dòng)態(tài)相關(guān)的狀態(tài)遷移需求增加，大狀態(tài)下的可擴(kuò)展性顯著降低。

5.3 Choosing Appropriate Parameters

?????? Elasticutor中有兩個(gè)重要參數(shù)：每個(gè)執(zhí)行程序的硬數(shù)，表示為z，每個(gè)運(yùn)算符的執(zhí)行數(shù)，表示為asy。作為一個(gè)規(guī)則，將z設(shè)置在256和1024之間可以實(shí)現(xiàn)良好的內(nèi)部執(zhí)行器負(fù)載平衡，并且將y設(shè)置為計(jì)算密集型運(yùn)營(yíng)商的節(jié)點(diǎn)數(shù)可以為那些運(yùn)營(yíng)商提供足夠的潛力來(lái)擴(kuò)展工作負(fù)載陣陣。然而，在下文中，我們?cè)诟鞣N工作負(fù)載下評(píng)估系統(tǒng)性能的大范圍（y，z），以便了解這兩個(gè)參數(shù)影響系統(tǒng)性能的原因和方式以及如何在極端情況下選擇合適的參數(shù)工作負(fù)載。為了進(jìn)行全面觀察，我們使用三種代表性工作負(fù)載，即默認(rèn)工作負(fù)載，數(shù)據(jù)密集型工作負(fù)載和高度動(dòng)態(tài)工作負(fù)載。設(shè)s和ω分別表示以字節(jié)為單位的元組大小和每分鐘的密鑰重組。在默認(rèn)工作負(fù)載中，（s，ω）=（128B，2）。我們分別通過(guò)將s增加到8K和ω增加到16來(lái)獲得數(shù)據(jù)密集型工作負(fù)載和高動(dòng)態(tài)工作負(fù)載。因此，（s，ω）=（8K，2）用于數(shù)據(jù)密集型工作負(fù)載，（s，ω）=（128B，16）用于高動(dòng)態(tài)工作負(fù)載。圖16顯示了在三個(gè)工作負(fù)載下具有各種y和z的系統(tǒng)吞吐量。為了比較，我們還在圖中顯示了靜態(tài)和RC方法的吞吐量。

?????? Number of shards：從圖16中，我們觀察到隨著z增加，吞吐量通常會(huì)增加，盡管邊際增長(zhǎng)正在減少。 這表明當(dāng)使用太少的分片時(shí)，例如，z≤64，執(zhí)行器內(nèi)負(fù)載平衡質(zhì)量差，妨礙彈性執(zhí)行器有效地利用多個(gè)核心; 然而，太精細(xì)的分片（例如，z≥1024）不會(huì)進(jìn)一步提高吞吐量，因?yàn)閳?zhí)行器內(nèi)的負(fù)載平衡已經(jīng)有效。 基于那些z = 16個(gè)觀測(cè)值，我們驗(yàn)證每個(gè)執(zhí)行器256到1024個(gè)分片實(shí)現(xiàn)了良好的性能。

?????? Number of executors：如圖16（a）所示，對(duì)于一個(gè)足夠大的z，除了y = 256之外，Elasticutor實(shí)現(xiàn)了有希望的性能。當(dāng)y = 256時(shí)，即集群中的CPU核心數(shù)量，每個(gè)彈性執(zhí)行器只能分配一個(gè)CPU核心。因此，執(zhí)行者失去彈性，Elasticutor被降級(jí)為靜態(tài)方法。通過(guò)比較圖16（a）和圖16（b），我們可以看到，當(dāng)元組大小增加到8K時(shí)，靜態(tài)和RC的性能變化不大，而在y = 1的情況下Elasticutor的性能?chē)?yán)重下降。與默認(rèn)工作負(fù)載相比，在數(shù)據(jù)密集型工作負(fù)載中運(yùn)行遠(yuǎn)程任務(wù)時(shí)遠(yuǎn)程數(shù)據(jù)傳輸?shù)某杀靖叱?4倍。這限制了單個(gè)執(zhí)行程序的可伸縮性，因此導(dǎo)致單個(gè)執(zhí)行程序需要擴(kuò)展到許多遠(yuǎn)程CPU核心的小y的性能較差。通過(guò)比較圖16（a）和圖16（c），我們觀察到隨著洗牌頻率從2增加到16，雖然吞吐量一般會(huì)減少，但當(dāng)y很小時(shí)，減少幅度要大得多，即1或者8.在頻繁混洗的動(dòng)態(tài)工作負(fù)載下，例如ω= 16，需要重新分配更多分片以進(jìn)行負(fù)載平衡，從而導(dǎo)致高遷移成本。相反，當(dāng)y足夠大時(shí)，大多數(shù)執(zhí)行器可以使用本地CPU內(nèi)核進(jìn)行擴(kuò)展，從而避免由于內(nèi)部處理狀態(tài)共享機(jī)制導(dǎo)致的狀態(tài)遷移;因此，吞吐量不會(huì)降低太多。總之，為每個(gè)節(jié)點(diǎn)設(shè)置一個(gè)或兩個(gè)執(zhí)行程序?qū)Ω鞣N工作負(fù)載都很穩(wěn)健。

5.4 Evaluation of Realtime Application

?????? 為了評(píng)估Elasticutor在實(shí)際應(yīng)用中的表現(xiàn)，我們使用上海證券交易所（SSE）交易的股票的匿名訂單數(shù)據(jù)集，收集時(shí)間超過(guò)三個(gè)月，每個(gè)交易時(shí)間約有800萬(wàn)條記錄。該應(yīng)用程序執(zhí)行證券交易所的市場(chǎng)清算機(jī)制，并提供實(shí)時(shí)分析。應(yīng)用程序的拓?fù)浣Y(jié)構(gòu)如圖17所示。輸入流由買(mǎi)方和賣(mài)方的限價(jià)訂單組成，這些限價(jià)訂單指定了特定庫(kù)存的特定交易量的出價(jià)和要價(jià)。順序元組的大小為96字節(jié)。在新訂單到達(dá)時(shí)，交易員操作員針對(duì)未完成的訂單執(zhí)行它，并確定交易數(shù)量和現(xiàn)金轉(zhuǎn)移。一旦進(jìn)行了這樣的交易，就會(huì)向下游運(yùn)營(yíng)商發(fā)送160字節(jié)的交易記錄，包括時(shí)間，股票數(shù)量和交易價(jià)格以及賣(mài)方，買(mǎi)方和股票的ID，包括6個(gè)運(yùn)營(yíng)商的統(tǒng)計(jì)數(shù)據(jù)和5個(gè)事件處理的運(yùn)營(yíng)商。分析運(yùn)算符生成統(tǒng)計(jì)數(shù)據(jù)，例如移動(dòng)平均值和綜合指數(shù)，并觸發(fā)用戶(hù)定義的事件，例如當(dāng)特定股票的交易價(jià)格超過(guò)預(yù)定義閾值時(shí)的警報(bào)。每個(gè)統(tǒng)計(jì)運(yùn)營(yíng)商的狀態(tài)大小約為200MB到400MB，而事件處理運(yùn)營(yíng)商的狀態(tài)相對(duì)較小，低于10MB。由于交易和分析涉及個(gè)股，我們將庫(kù)存ID的空間劃分為并行處理。由于股票交易具有不可預(yù)測(cè)的性質(zhì)，股票的到達(dá)率和分布均隨著時(shí)間的推移而波動(dòng)很大，從而導(dǎo)致高度動(dòng)態(tài)的工作量。為了說(shuō)明工作負(fù)荷動(dòng)態(tài)，圖18顯示了5種最受歡迎??的股票的即時(shí)到達(dá)率。

?????? 除了靜態(tài)，RC和Elasticutor之外，我們還測(cè)試了一個(gè)天真的以執(zhí)行器為中心（naive-EC）的實(shí)現(xiàn)，它與Elasticutor相同，只是在調(diào)度程序中禁用了遷移成本和計(jì)算局部性的優(yōu)化。 圖19繪制了在32個(gè)節(jié)點(diǎn)上運(yùn)行的四種方法下的瞬時(shí)吞吐量和第99百分位處理延遲。 我們觀察到，naive-EC和Elasicutor都優(yōu)于靜態(tài)和RC方法，大約使吞吐量翻倍，并將延遲降低1-2個(gè)數(shù)量級(jí)。 盡管naive-EC和Elasticutor之間的性能差距是可識(shí)別的，但與執(zhí)行者為中心的方法和其他兩種方法之間的差距相比，它們之間的差距很小。 這一觀察結(jié)果表明，盡管動(dòng)態(tài)調(diào)度程序中的優(yōu)化能夠顯著提高性能，但Elasticutor的更好性能主要?dú)w功于采用的以執(zhí)行器為中心的有利范式。

?????? 為了進(jìn)一步說(shuō)明naive-EC和Elasticutor之間性能差距背后的原因，我們?cè)诒?中顯示了它們的狀態(tài)遷移率和遠(yuǎn)程數(shù)據(jù)傳輸率。前者的速率是整個(gè)系統(tǒng)在網(wǎng)絡(luò)中遷移的狀態(tài)的聚合大小。 單位時(shí)間。 后一種速率是在所有彈性執(zhí)行器和它們的遠(yuǎn)程任務(wù)之間以單位時(shí)間傳輸?shù)臄?shù)據(jù)的總量。 我們觀察到，naive-EC下的狀態(tài)遷移率和遠(yuǎn)程數(shù)據(jù)傳輸率分別比Elasticutor下的5倍和10倍高。 通過(guò)較少的狀態(tài)遷移，彈性執(zhí)行器轉(zhuǎn)換到新的資源分配計(jì)劃將更有效，從而實(shí)現(xiàn)更高的性能。 同樣，通過(guò)較少的遠(yuǎn)程數(shù)據(jù)傳輸，Operator間數(shù)據(jù)傳輸可以使用更多的網(wǎng)絡(luò)帶寬，從而進(jìn)一步提高性能。

?????? 最后，我們?cè)赟SE工作負(fù)載下評(píng)估Elasticutor的可伸縮性。 我們改變計(jì)算集群的大小，即節(jié)點(diǎn)的數(shù)量，并測(cè)量Elasticutor的吞吐量和調(diào)度成本，即動(dòng)態(tài)調(diào)度器計(jì)算新的CPU到執(zhí)行器分配所需的平均時(shí)間。 保持較低的調(diào)度成本對(duì)于系統(tǒng)適應(yīng)動(dòng)態(tài)工作負(fù)載非常重要。 表3顯示了隨著規(guī)模增加的吞吐量和調(diào)度成本。 我們觀察到隨著集群的增長(zhǎng)，吞吐量幾乎呈線性增長(zhǎng); 并且調(diào)度成本大約是幾毫秒，并且隨著節(jié)點(diǎn)數(shù)量的增加而略有增長(zhǎng).

6. RELATEDWORK

?????? Stream Processing System 早期的流處理系統(tǒng)，如Aurora [7]，Borealis [6]，TelegraphCQ [17]和STREAM [10]，旨在通過(guò)利用分布式但靜態(tài)的計(jì)算資源來(lái)處理海量數(shù)據(jù)更新。 借助云計(jì)算技術(shù)，出現(xiàn)了新一代流系統(tǒng)，重點(diǎn)是并行數(shù)據(jù)處理，可用性和容錯(cuò)，以充分利用基于云的平臺(tái)上的靈活資源管理方案。 Spark Streaming [50]，Storm [43]，Samza [35]，Heron [31]，Flink [12]和Waterwheel [46]是最流行的開(kāi)源系統(tǒng)，提供分布式流處理和分析。 大型工業(yè)企業(yè)也在開(kāi)發(fā)內(nèi)部分布式流系統(tǒng)，如Muppet [32]，MillWheel [8]，Trill [16]，Dataflow [9]和StreamScope [33]。

Elasticity.大量的工作探索了實(shí)現(xiàn)彈性的可能性。卡斯特羅等人。 [15]將資源重新擴(kuò)展操作與分布式流系統(tǒng)中的容錯(cuò)功能相結(jié)合，以便在遷移到新計(jì)算節(jié)點(diǎn)之前將與處理邏輯綁定的中間狀態(tài)寫(xiě)入持久性存儲(chǔ)。王等人。 [47]提出了彈性流水線技術(shù)，以便為分布式SQL查詢(xún)啟用動(dòng)態(tài)，工作負(fù)載感知的運(yùn)行時(shí)重新配置。在Flux [39]中提出了一種自適應(yīng)分區(qū)算子，以實(shí)現(xiàn)節(jié)點(diǎn)之間的分區(qū)移動(dòng)以實(shí)現(xiàn)負(fù)載平衡。但是，由于其工作負(fù)載遷移基于每個(gè)分區(qū)，因此當(dāng)單個(gè)分區(qū)超出群集中任何節(jié)點(diǎn)的處理能力時(shí)，此方法將面臨困難。 ChronoStream [48]將計(jì)算狀態(tài)劃分為一個(gè)集合的粒度切片單元，并在節(jié)點(diǎn)之間動(dòng)態(tài)分配它們以支持彈性。 Gedik等人。 [24]提出了在不違反狀態(tài)一致性的情況下擴(kuò)展有狀態(tài)運(yùn)算符的機(jī)制。 Chi [34]是一個(gè)具有監(jiān)控和動(dòng)態(tài)重新配置功能的控制面板。然而，這些方法在以資源為中心的標(biāo)準(zhǔn)之后實(shí)現(xiàn)了彈性，這導(dǎo)致了昂貴的同步并且妨礙了快速?gòu)椥浴Ｋ鼈冞m用于以粗糙時(shí)間粒度使用彈性功能的情況，即每5分鐘;實(shí)現(xiàn)的彈性太慢，無(wú)法應(yīng)用于具有高動(dòng)態(tài)工作負(fù)載的應(yīng)用中。 Elasticutor采用新的以執(zhí)行器為中心的方法來(lái)避免這個(gè)問(wèn)題。此方法大大降低了執(zhí)行工作負(fù)載重新同步的同步開(kāi)銷(xiāo)，因此可在幾毫秒內(nèi)實(shí)現(xiàn)工作負(fù)載重新分配。

Workload Distribution。分布式流系統(tǒng)的通用工作負(fù)載分配是一個(gè)具有挑戰(zhàn)性的問(wèn)題，因?yàn)殡S著時(shí)間的推移，輸入數(shù)據(jù)流的偏差很大并且差異很大。 Shah [39]等。為傳統(tǒng)流處理框架中的單個(gè)操作設(shè)計(jì)了動(dòng)態(tài)工作負(fù)載再分配機(jī)制，例如Borealis [6]。 [24]和[20]研究了混合路由策略，通過(guò)其密鑰對(duì)工作負(fù)載進(jìn)行分組，以便根據(jù)CPU，內(nèi)存和帶寬資源動(dòng)態(tài)平衡負(fù)載。 TimeStream [38]采用圖形重組策略，直接用全新的處理拓?fù)涮鎿Q原始處理拓?fù)洹Ｈ欢?#xff0c;系統(tǒng)在所有可應(yīng)用的圖形結(jié)構(gòu)的巨大搜索空間中監(jiān)視和優(yōu)化拓?fù)浣Y(jié)構(gòu)具有挑戰(zhàn)性。 Cardellini等。 [13]研究了Storm之上的有狀態(tài)任務(wù)遷移。丁等人。 [18]討論了基于馬爾可夫決策過(guò)程（MDP）制定任務(wù)遷移計(jì)劃的長(zhǎng)期優(yōu)化，以提高分布式流引擎的資源利用率。但是，Elasticutor不僅可以實(shí)現(xiàn)工作負(fù)載分配中的負(fù)載平衡，還可以考慮遷移成本最小化和計(jì)算局部性。

7. Conclusion

我們?cè)O(shè)計(jì)并實(shí)現(xiàn)了Elasticutor，它為流處理系統(tǒng)提供了快速?gòu)椥浴?彈性器遵循一種新的以執(zhí)行器為中心的方法，該方法將執(zhí)行程序靜態(tài)綁定到運(yùn)算符，但允許執(zhí)行程序獨(dú)立擴(kuò)展。 這種方法將操作員的擴(kuò)展與有狀態(tài)處理所需的全局同步分離開(kāi)來(lái)。 Elasticutor框架有兩個(gè)構(gòu)建塊：彈性執(zhí)行器，執(zhí)行動(dòng)態(tài)負(fù)載平衡，以及優(yōu)化計(jì)算資源使用的調(diào)度程序。 實(shí)驗(yàn)表明，與傳統(tǒng)的以資源為中心的提供彈性的方法相比，彈性器使吞吐量增加一倍，平均延遲降低了幾個(gè)數(shù)量級(jí)。

總結(jié)

以上是生活随笔為你收集整理的一周一论文（翻译）——[SIGMOD 19] Elasticutor：Rapid Elasticity for Realtime Stateful Stream Processing的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。