作者：Stephan Ewen (Apache Flink最早的創(chuàng)建者之一) 整理：秦江杰 Flink社區(qū)?？螺d地址 第一期：不僅僅是流計算：https://files.alicdn.com/tpsservice/a23b4f3a443c9977a39fb806c30d96fd.pdf 第二期：重新定義計算：https://files.alicdn.com/tpsservice/2808df944bfc81048bd7a543123ff437.pdf 本文整理自Stephan Ewen在Flink Forward China 2018 上的演講《Stream Processing takes on Everything》。

大家好，今天我的演講主題看似比較激進：流處理解決所有問題。從某種程度上來說，你可以認為這個演講是之前曉偉對Flink未來(詳情參見上一篇文章：Apache Flink?- 重新定義計算)描述的一個繼續(xù)。很多人可能對Flink還是停留在最初的認知，覺得Flink是一個流處理引擎，實際上Flink可以做很多其他的工作，比如批處理，比如應用程序。接下來我會簡單的說明我對Flink功能的觀點，然后我會深入介紹一個特別領域的應用和事件處理場景。這個場景乍看起來不是一個流處理的使用場景，但是在我看來，實際上它就是一個很有趣的流處理使用場景。

上圖對為什么流處理可以處理一切作出詮釋，將數(shù)據(jù)看做流是一個自然而又十分強大的想法。大部分數(shù)據(jù)的產生過程都是隨時間生成的流，比如一個Petabyte的數(shù)據(jù)不會憑空產生。這些數(shù)據(jù)通常都是一些事件的積累，比如支付、將商品放入購物車，網頁瀏覽，傳感器采樣輸出。 基于數(shù)據(jù)是流的想法，我們對數(shù)據(jù)處理可以有相應的理解。比如將過去的歷史數(shù)據(jù)看做是一個截止到某一時刻的有限的流，或是將一個實時處理應用看成是從某一個時刻開始處理未來到達的數(shù)據(jù)。可能在未來某個時刻它會停止，那么它就變成了處理從開始時刻到停止時刻的有限數(shù)據(jù)的批處理。當然，它也有可能一直運行下去，不斷處理新到達的數(shù)據(jù)。這個對數(shù)據(jù)的重要理解方式非常強大，基于這一理解，F(xiàn)link可以支持整個數(shù)據(jù)處理范疇內的所有場景。

最廣為人知的Flink使用場景是流分析、連續(xù)處理(或者說漸進式處理)，這些場景中Flink實時或者近實時的處理數(shù)據(jù)，或者采集之前提到的歷史數(shù)據(jù)并且連續(xù)的對這些事件進行計算。曉偉在之前的演講中提到一個非常好的例子來說明怎么樣通過對Flink進行一些優(yōu)化，進而可以針對有限數(shù)據(jù)集做一些特別的處理，這使得Flink能夠很好的支持批處理的場景，從性能上來說能夠與最先進的批處理引擎相媲美。而在這根軸的另一頭，是我今天的演講將要說明的場景 – 事件驅動的應用。這類應用普遍存在于任何服務或者微服務的架構中。這類應用接收各類事件(可能是RPC調用、HTTP請求)，并且對這些事件作出一些響應，比如把商品放進購物車，或者加入社交網絡中的某個群組。

在我進一步展開今天的演講之前，我想先對社區(qū)在Flink的傳統(tǒng)領域(實時分析、連續(xù)處理)近期所做的工作做一個介紹。Flink 1.7在2018年11月30日已經發(fā)布。在Flink 1.7中為典型的流處理場景加入了一些非常有趣的功能。比如我個人非常感興趣的在流式SQL中帶時間版本的Join。一個基本想法是有兩個不同的流，其中一個流被定義為隨時間變化的參照表，另一個是與參照表進行Join的事件流。比如事件流是一個訂單流，參照表是不斷被更新的匯率，而每個訂單需要使用最新的匯率來進行換算，并將換算的結果輸出到結果表。這個例子在標準的SQL當中實際上并不容易表達，但在我們對Streaming SQL做了一點小的擴展以后，這個邏輯表達變得非常簡單，我們發(fā)現(xiàn)這樣的表達有非常多的應用場景。 另一個在流處理領域十分強大的新功能是將復雜事件處理(CEP)和SQL相結合。CEP應用觀察事件模式。比如某個CEP應用觀察股市，當有兩個上漲后緊跟一個下跌時，這個應用可能做些交易。再比如一個觀察溫度計的應用，當它發(fā)現(xiàn)有溫度計在兩個超過90攝氏度的讀數(shù)之后的兩分鐘里沒有任何操作，可能會進行一些操作。與SQL的結合使這類邏輯的表達也變得非常簡單。 第三個Flink 1.7中做了很多工作的功能是Schema升級。這個功能和基于流的應用緊密相關。就像你可以對數(shù)據(jù)庫進行數(shù)據(jù)Schema升級一樣，你可以修改Flink表中列的類型或者重新寫一個列。 另外我想簡單介紹的是流處理技術不僅僅是簡單對數(shù)據(jù)進行計算，這還包括了很多與外部系統(tǒng)進行事務交互。流處理引擎需要在采用不同協(xié)議的系統(tǒng)之間以事務的方式移動數(shù)據(jù)，并保證計算過程和數(shù)據(jù)的一致性。這一部分功能也是在Flink 1.7中得到了增強。

以上我對Flink 1.7的新功能向大家做了簡單總結。下面讓我們來看看今天我演講的主要部分，也就是利用Flink來搭建應用和服務。我將說明為什么流處理是一個搭建應用和服務或者微服務的有趣技術。

我將從左邊這個高度簡化的圖說起，我們一會兒將聊一些其中的細節(jié)。首先我們來看一個理解應用簡單的視角。如左圖所示，一個應用可以是一個Container，一個Spring應用，或者Java應用、Ruby應用，等等。這個應用從諸如RPC，HTTP等渠道接收請求，然后依據(jù)請求進行數(shù)據(jù)庫變更。這個應用也可能調用另一個微服務并進行下一步的處理。我們可以非常自然的想到進入到應用的這些請求可以看做是個事件組成的序列，所以我們可以把它們看做是事件流。可能這些事件被緩存在消息隊列中，而應用會從消息隊列中消費這些事件進行處理，當應用需要響應一個請求時，它將結果輸出到另一個消息隊列，而請求發(fā)送方可以從這個消息隊列中消費得到所發(fā)送請求的響應。在這張圖中我們已經可以看到一些有趣的不同。 第一個不同是在這張圖中應用和數(shù)據(jù)庫不再是分開的兩個實體，而是被一個有狀態(tài)的流處理應用所代替。所以在流處理應用的架構中，不再有應用和數(shù)據(jù)庫的連接了，它們被放到了一起。這個做法有利有弊，但其中有些好處是非常重要的。首先是性能上的好處是明顯的，因為應用不再需要和數(shù)據(jù)庫進行交互，處理可以基于內存中的變量進行。其次這種做法有很好并且很簡單的一致性。

這張圖被簡化了很多，實際上我們通常會有很多個應用，而不是一個被隔離的應用，很多情況下你的應用會更符合這張圖。系統(tǒng)中有個接收請求的接口，然后請求被發(fā)送到第一個應用，可能會再被發(fā)到另一個應用，然后得到相應。在圖中有些應用會消費中間結果的流。這張圖已經展示了為什么流處理是更適合比較復雜的微服務場景的技術。因為很多時候系統(tǒng)中不會有一個直接接收用戶請求并直接響應的服務，通常來說一個微服務需要跟其他微服務通信。這正如在流處理的架構中不同應用在創(chuàng)建輸出流，同時基于衍生出的流再創(chuàng)建并輸出新的流。

到目前為止，我們看到的內容多少還比較直觀。而對基于流處理技術的微服務架構而言，人們最常問的一個問題是如何保證事務性?如果系統(tǒng)中使用的是數(shù)據(jù)庫，通常來說都會有非常成熟復雜的數(shù)據(jù)校驗和事務模型。這也是數(shù)據(jù)庫在過去許多年中十分成功的原因。開始一個事務，對數(shù)據(jù)做一些操作，提交或者撤銷一個事務。這個機制使得數(shù)據(jù)完整性得到了保證(一致性，持久性等等)。 那么在流處理中我們怎么做到同樣的事情呢?作為一個優(yōu)秀的流處理引擎，F(xiàn)link支持了恰好一次語義，保證了每個事件只會被處理一遍。但是這依然對某些操作有限制，這也成為了使用流處理應用的一個障礙。我們通過一個非常簡單流處理應用例子來看我們可以做一些什么擴展來解決這個問題。我們會看到，解決辦法其實出奇的簡單。

讓我們以這個教科書式的事務為例子來看一下事務性應用的過程。這個系統(tǒng)維護了賬戶和其中存款余額的信息。這樣的信息可能是銀行或者在線支付系統(tǒng)的場景中用到的。假設我們想要處理類似下面的事務： 如果賬戶A中的余額大于100，那么從賬戶A中轉賬50元到賬戶B。 這是個非常簡單的兩個賬戶之間進行轉賬的例子。 數(shù)據(jù)庫對于這樣的事務已經有了一個核心的范式，也就是原子性，一致性，隔離性和持久性(ACID)。這是能夠讓用戶放心使用事務的幾個基本保證。有了他們，用戶不用擔心錢在轉賬過程中會丟失或者其他問題。讓我們用這個例子來放到流處理應用中，來讓流處理應用也能提供和數(shù)據(jù)相同的ACID支持： 原子性要求一個轉賬要不就完全完成，也就是說轉賬金額從一個賬戶減少，并增加到另一個賬戶，要不就兩個賬戶的余額都沒有變化。而不會只有一個賬戶余額改變。否則的話錢就會憑空減少或者憑空增加。 一致性和隔離性是說如果有很多用戶同時想要進行轉賬，那么這些轉賬行為之間應該互不干擾，每個轉賬行為應該被獨立的完成，并且完成后每個賬戶的余額應該是正確的。也就是說如果兩個用戶同時操作同一個賬戶，系統(tǒng)不應該出錯。 持久性指的是如果一個操作已經完成，那么這個操作的結果會被妥善的保存而不會丟失。 我們假設持久性已經被滿足。一個流處理器有狀態(tài)，這個狀態(tài)會被checkpoint，所以流處理器的狀態(tài)是可恢復的。也就是說只要我們完成了一個修改，并且這個修改被checkpoint了，那么這個修改就是持久化的。

讓我們來看看另外三個例子。設想一下，如果我們用流處理應用來實現(xiàn)這樣一個轉賬系統(tǒng)會發(fā)生什么。我們先把問題簡化一些，假設轉賬不需要有條件，僅僅是將50元從賬戶A轉到賬戶，也就是說賬戶A的余額減少50元而賬戶B的余額增加50元。我們的系統(tǒng)是一個分布式的并行系統(tǒng)，而不是一個單機系統(tǒng)。簡單起見我們假設系統(tǒng)中只有兩臺機器，這兩臺機器可以是不同的物理機或者是在YARN或者Kubernetes上不同的容器?？傊鼈兪莾蓚€不同的流處理器實例，數(shù)據(jù)分布在這兩個流處理器上。我們假設賬戶A的數(shù)據(jù)由其中一臺機器維護，而賬戶B的數(shù)據(jù)有另一臺機器維護。 現(xiàn)在我們要做個轉賬，將50元從賬戶A轉移到賬戶B，我們把這個請求放進隊列中，然后這個轉賬請求被分解為對賬戶A和B分別進行操作，并且根據(jù)鍵將這兩個操作路由到維護賬戶A和維護賬戶B的這兩臺機器上，這兩臺機器分別根據(jù)要求對賬戶A和賬戶B的余額進行改動。這并不是事務操作，而只是兩個獨立無意義的改動。一旦我們將轉賬的請求改的稍微復雜一些就會發(fā)現(xiàn)問題。

下面我們假設轉賬是有條件的，我們只想在賬戶A的余額足夠的情況下才進行轉賬，這樣就已經有些不太對了。如果我們還是像之前那樣操作，將這個轉賬請求分別發(fā)送給維護賬戶A和B的兩臺機器，如果A沒有足夠的余額，那么A的余額不會發(fā)生變化，而B的余額可能已經被改動了。我們就違反了一致性的要求。 我們看到我們需要首先以某種方式統(tǒng)一做出是否需要更改余額的決定，如果這個統(tǒng)一的決定中余額需要被修改，我們再進行修改余額的操作。所以我們先給維護A的余額的機器發(fā)送一個請求，讓它查看A的余額。我們也可以對B做同樣的事情，但是這個例子里面我們不關心B的余額。然后我們把所有這樣的條件檢查的請求匯總起來去檢驗條件是否滿足。因為Flink這樣的流處理器支持迭代，如果滿足轉賬條件，我們可以把這個余額改動的操作放進迭代的反饋流當中來告訴對應的節(jié)點來進行余額修改。反之如果條件不滿足，那么余額改動的操作將不會被放進反饋流。這個例子里面，通過這種方式我們可以正確的進行轉賬操作。從某種角度上來說我們實現(xiàn)了原子性，基于一個條件我們可以進行全部的余額修改，或者不進行任何余額修改。這部分依然還是比較直觀的，更大的困難是在于如何做到并發(fā)請求的隔離性。

假設我們的系統(tǒng)沒有變，但是系統(tǒng)中有多個并發(fā)的請求。我們在之前的演講中已經知道，這樣的并發(fā)可能達到每秒鐘幾十億條。如圖，我們的系統(tǒng)可能從兩個流中同時接受請求。如果這兩個請求同時到達，我們像之前那樣將每個請求拆分成多個請求，首先檢查余額條件，然后進行余額操作。然而我們發(fā)現(xiàn)這會帶來問題。管理賬戶A的機器會首先檢查A的余額是否大于50，然后又會檢查A的余額是否大于100，因為兩個條件都滿足，所以兩筆轉賬操作都會進行，但實際上賬戶A上的余額可能無法同時完成兩筆轉賬，而只能完成50元或者100元的轉賬中的一筆。這里我們需要進一步思考怎么樣來處理并發(fā)的請求，我們不能只是簡單地并發(fā)處理請求，這會違反事務的保證。從某種角度來說，這是整個數(shù)據(jù)庫事務的核心。數(shù)據(jù)庫的專家們花了一些時間提供了不同解決方案，有的方案比較簡單，有的則很復雜。但所有的方案都不是那么容易，尤其是在分布式系統(tǒng)當中。 在流處理中怎么解決這個問題呢?直覺上講，如果我們能夠讓所有的事務都按照順序依次發(fā)生，那么問題就解決了，這也被成為可序列化的特性。但是我們當然不希望所有的請求都被依次順序處理，這與我們使用分布式系統(tǒng)的初衷相違背。所以我們需要保證這些請求最后的產生的影響看起來是按照順序發(fā)生的，也就是一個請求產生的影響是基于前一個請求產生影響的基礎之上的。換句話說也就是一個事務的修改需要在前一個事務的所有修改都完成后才能進行。這種希望一件事在另一件事之后發(fā)生的要求看起來很熟悉，這似乎是我們以前在流處理中曾經遇到過的問題。是的，這聽上去像是事件時間。用高度簡化的方式來解釋，如果所有的請求都在不同的事件時間產生，即使由于種種原因他們到達處理器的時間是亂序的，流處理器依然會根據(jù)他們的事件時間來對他們進行處理。流處理器會使得所有的事件的影響看上去都是按順序發(fā)生的。按事件時間處理是Flink已經支持的功能。

那么詳細說來，我們到底怎么解決這個一致性問題呢?假設我們有并行的請求輸入并行的事務請求，這些請求讀取某些表中的記錄，然后修改某些表中的記錄。我們首先需要做的是把這些事務請求根據(jù)事件時間順序擺放。這些請求的事務時間不能夠相同，但是他們之間的時間也需要足夠接近，這是因為在事件時間的處理過程中會引入一定的延遲，我們需要保證所處理的事件時間在向前推進。因此第一步是定義事務執(zhí)行的順序，也就是說需要有一個聰明的算法來為每個事務制定事件時間。在圖上，假設這三個事務的事件時間分別是T+2, T和T+1。那么第二個事務的影響需要在第一和第三個事務之前。不同的事務所做的修改是不同的，每個事務都會產生不同的操作請求來修改狀態(tài)。我們現(xiàn)在需要將對訪問每個行和狀態(tài)的事件進行排序，保證他們的訪問是符合事件時間順序的。這也意味著那些相互之間沒有關系的事務之間自然也沒有了任何影響。比如這里的第三個事務請求，它與前兩個事務之間沒有訪問共同的狀態(tài)，所以它的事件時間排序與前兩個事務也相互獨立。而當前兩個事務之間的操作的到達順序與事件時間不符時，F(xiàn)link則會依據(jù)它們的事件時間進行排序后再處理。 必須承認，這樣說還是進行了一些簡化，我們還需要做一些事情來保證高效執(zhí)行，但是總體原則上來說，這就是全部的設計。除此之外我們并不需要更多其他東西。

為了實現(xiàn)這個設計，我們引入了一種聰明的分布式事件時間分配機制。這里的事件時間是邏輯時間，它并不需要有什么現(xiàn)實意義，比如它不需要是真實的時鐘。使用Flink的亂序處理能力，并且使用Flink迭代計算的功能來進行某些前提條件的檢查。這些就是我們構建一個支持事務的流處理器的要素。

我們實際上已經完成了這個工作，稱之為流式賬簿(Streaming Ledger)，這是個在Apache Flink上很小的庫。它基于流處理器做到了滿足ACID的多鍵事務性操作。我相信這是個非常有趣的進化。流處理器一開始基本上沒有任何保障，然后類似Storm的系統(tǒng)增加了至少一次的保證。但顯然至少一次依然不夠好。然后我們看到了恰好一次的語義，這是一個大的進步，但這只是對于單行操作的恰好一次語義，這與鍵值庫很類似。而支持多行恰好一次或者多行事務操作將流處理器提升到了一個可以解決傳統(tǒng)意義上關系型數(shù)據(jù)庫所應用場景的階段。

Streaming Ledger的實現(xiàn)方式是允許用戶定義一些表和對這些表進行修改的函數(shù)。Streaming Ledger會運行這些函數(shù)和表，所有的這些一起編譯成一個Apache Flink的有向無環(huán)圖(DAG)。Streaming Ledger會注入所有事務時間分配的邏輯，以此來保證所有事務的一致性。

搭建這樣一個庫并不難，難的是讓它高性能的運行。讓我們來看看它的性能。這些性能測試是幾個月之前的，我們并沒有做什么特別的優(yōu)化，我們只是想看看一些最簡單的方法能夠有什么樣的性能表現(xiàn)。而實際性能表現(xiàn)看起來相當不錯。如果你看這些性能條形成的階梯跨度，隨著流處理器數(shù)量的增長，性能的增長相當線性。在事務設計中，沒有任何協(xié)同或者鎖參與其中。這只是流處理，將事件流推入系統(tǒng)，緩存一小段時間來做一些亂序處理，然后做一些本地狀態(tài)更新。在這個方案中，沒有什么特別代價高昂的操作。在圖中性能增長似乎超過了線性，我想這主要是因為JAVA的JVM當中GC的工作原因導致的。在32個節(jié)點的情況下我們每秒可以處理大約兩百萬個事務。為了與數(shù)據(jù)庫性能測試進行對比，通常當你看數(shù)據(jù)庫的性能測試時，你會看到類似讀寫操作比的說明，比如10%的更新操作。而我們的測試使用的是100%的更新操作，而每個寫操作至少更新在不同分區(qū)上的4行數(shù)據(jù)，我們的表的大小大約是兩億行。即便沒有任何優(yōu)化，這個方案的性能也非常不錯。 另一個在事務性能中有趣的問題是當更新的操作對象是一個比較小的集合時的性能。如果事務之間沒有沖突，并發(fā)的事務處理是一個容易的事情。如果所有的事務都獨立進行而互不干擾，那這個不是什么難題，任何系統(tǒng)應該都能很好的解決這樣的問題。當所有的事務都開始操作同一些行時，事情開始變得更有趣了，你需要隔離不同的修改來保證一致性。所以我們開始比較一個只讀的程序、一個又讀又寫但是沒有寫沖突的程序和一個又讀又寫并有中等程度寫沖突的程序這三者之間的性能。你可以看到性能表現(xiàn)相當穩(wěn)定。這就像是一個樂觀的并發(fā)沖突控制，表現(xiàn)很不錯。那如果我們真的想要針對這類系統(tǒng)的阿喀琉斯之踵進行考驗，也就是反復的更新同一個小集合中的鍵。在傳統(tǒng)數(shù)據(jù)庫中，這種情況下可能會出現(xiàn)反復重試，反復失敗再重試，這是一種我們總想避免的糟糕情況。是的，我們的確需要付出性能代價，這很自然，因為如果你的表中有幾行數(shù)據(jù)每個人都想更新，那么你的系統(tǒng)就失去了并發(fā)性，這本身就是個問題。但是這種情況下，系統(tǒng)并沒崩潰，它仍然在穩(wěn)定的處理請求，雖然失去了一些并發(fā)性，但是請求依然能夠被處理。這是因為我們沒有沖突重試的機制，你可以認為我們有一個基于亂序處理天然的沖突避免的機制，這是一種非常穩(wěn)定和強大的技術。

我們還嘗試了在跨地域分布的情況下的性能表現(xiàn)。比如我們在美國、巴西，歐洲，日本和澳大利亞各設置了一個Flink集群。也就是說我們有個全球分布的系統(tǒng)。如果你在使用一個關系型數(shù)據(jù)庫，那么你會付出相當高昂的性能代價，因為通信的延遲變得相當高?？绱笾薜男畔⒔换ケ仍谕粋€數(shù)據(jù)中心甚至同一個機架上的信息交互要產生大得多的延遲。但是有趣的是，流處理的方式對延遲并不是十分敏感，延遲對性能有所影響，但是相比其它很多方案，延遲對流處理的影響要小得多。所以，在這樣的全球分布式環(huán)境中執(zhí)行分布式程序，的確會有更差的性能，部分原因也是因為跨大洲的通信帶寬不如統(tǒng)一數(shù)據(jù)中心里的帶寬，但是性能表現(xiàn)依然不差。實際上，你可以拿它當做一個跨地域的數(shù)據(jù)庫，同時仍然能夠在一個大概10個節(jié)點的集群上獲得每秒幾十萬條事務的處理能力。在這個測試中我們只用了10個節(jié)點，每個大洲兩個節(jié)點。所以10個節(jié)點可以帶來全球分布的每秒20萬事務的處理能力。我認為這是很有趣的結果，這是因為這個方案對延遲并不敏感。

我已經說了很多利用流處理來實現(xiàn)事務性的應用。可能聽起來這是個很自然的想法，從某種角度上來說的確是這樣。但是它的確需要一些很復雜的機制來作為支撐。它需要一個連續(xù)處理而非微批處理的能力，需要能夠做迭代，需要復雜的基于事件時間處理亂序處理。為了更好地性能，它需要靈活的狀態(tài)抽象和異步checkpoint機制。這些是真正困難的事情。這些不是由Ledger Streaming庫實現(xiàn)的，而是Apache Flink實現(xiàn)的，所以即使對這類事務性的應用而言，Apache Flink也是真正的中流砥柱。

至此，我們可以說流處理不僅僅支持連續(xù)處理、流式分析、批處理或者事件驅動的處理，你也可以用它做事務性的處理。當然，前提是你有一個足夠強大的流處理引擎。這就是我演講的全部內容。 --------------------------------------------------------- 免責聲明： 1.本文援引自互聯(lián)網，旨在傳遞更多網絡信息，僅代表作者本人觀點，與本網站無關。 2.本文僅供讀者參考，本網站未對該內容進行證實，對其原創(chuàng)性、真實性、完整性、及時性不作任何保證。