更多深度文章，請關注雲計算頻道：https://yq.aliyun.com/cloud
阿里雲數據事業部強琦為大家帶來題為“流式計算的系統設計與實現”的演講，本文主要從增量計算和流式計算開始談起，然後講解了與批量計算的區別，重點對典型系統技術概要進行了分析，包括Storm、Kinesis、MillWheel，接著介紹了核心技術、消息機制以及StreamSQL等，一起來了解下吧。
增量計算和流式計算
流式計算
流計算對於時效性要求比較嚴格，實時計算就是對計算的時效性要求比較強。流計算是利用分佈式的思想和方法，對海量“流”式數據進行實時處理的系統，它源自對海量數據“時效”價值上的挖掘訴求。
那麼，通常說的實時系統或者實時計算，嚴格意義上來說分成三大類：
ad-hoc computing（數據的實時計算）：計算不可枚舉，計算在query時發生。
stream computing（實時數據的計算）：計算可枚舉，計算在數據發生變化時發生。
continuous computing(實時數據的實時計算)：大數據集的在線複雜實時計算。
增量計算
增量計算是分批，也就是batch，每個batch會計算出一個function的delta值，數據的一個delta最終會變成對function的一個delta值，最終通過增量計算達到效果。
batch => delta: f(x + delta) = g( f(x), delta )
實際上是在數據的delta值上計算的一個結果，這個f（x）我們稱之為oldValue，整個function的一個oldValue從公式就可以看到，整個增量計算與全量計算和批量計算有很大的不一樣的地方，就在於它是有狀態的計算，而批量計算系統和全量計算系統是無狀態的計算，所以這就會導致整個系統的設計思路理念和整個的容錯機制會有很大的不同，相對於oldValue本批次的數據，delta作為一個輸入，整體上是一個有狀態的計算，它會在系統的時效性、系統的複雜性和系統性能之間去做tradeoff，如果batch裡的數據量是非常少的，那這個系統表現出來的時效性是最實時的，當然，整個系統的容錯吞吐就會受到影響，就是說一批次的數據量是比較少的情況下，整個的系統吞吐會比較低，整個系統的容錯複雜度也會比較高，那麼在增量計算情況下，它有哪些優勢呢？
1.相比以前的全量計算，中間的計算結果是實時產出的，也就是說它的時效性是很強的；
2.我們把一個計算平攤在每一個時間段，可以做到平攤計算。整個集群的規模是受峰值的影響，雙十一的峰值流量是非常大的，如果按照最峰值的流量去計算，整個服務器資源是相對較高的，如果能夠把傳統的計算平攤在每一分鐘每一秒，實際可以起到降低成本的作用；
3.整個數據處理鏈路如果放在一次Query中進行處理，也即是全部的數據在進行一個function的計算時，會大量膨脹中間結果，也就是說像Group By Count會到達200G，而增量計算可以做到中間結果不膨脹；
4.增量計算是一個有狀態的計算，在分佈式領域，有狀態的failover策略會跟無狀態的計算系統截然不同，但是它的優勢是恢復快，任務可以切成很多碎片去運行，一旦任務因為任何幾臺服務器的抖動而宕機，整個的恢復是從前一次有效的batch開始計算，而不是像全量計算和離線計算一樣，全部要重新進行計算，當在離線計算和在線計算混合部署的情況下，這顯得尤為重要；
5.增量計算把一大塊數據分批去計算，因此在批量計算裡面經常遇到會一些數據傾斜問題在增量計算並不會遇到。在真實場景下，數據傾斜會對整個計算系統產生非常致命的影響，所以假設不同的節點之間數據傾斜比是1000，這個實際是很平常的，雙十一的時候，光小米一家店鋪就做到了很高的銷售額，小米店鋪和其他店鋪的成交是上萬倍甚至幾十萬倍的scale，傳統的分佈式計算的整個計算延時是受最慢的那個節點影響，如果把全部的數據分批次，實際上對於每一批來說，數據的傾斜度就會緩解，而且每個批次是可以並行去運行的，所以這可以大大地去降低整個計算任務在數據傾斜情況下的運行效率問題。
增量計算和流式計算應用場景
日誌採集和在線分析：如基於訪問日誌、交易數據的BI算法分析。比較有名的像Google的統計、百度的統計,一些網站根據訪問日誌，會分析出各種的UV、 PV、 IPV等運營指標，有了流式計算，就可以對這些訪問的時效性做到秒級、分鐘級的監控，比如雙十一當天，不同的店鋪會通過店鋪的實時訪問情況來決定後面的運營策略；
大數據的預處理：數據清洗、字段補全等；
風險監測與告警：如交易業務的虛假交易實時監測與分析；
網站與移動應用分析統計：如雙11運營、淘寶量子統計、CNZZ、友盟等各類統計業務；
網絡安全監測：如CDN的惡意攻擊分析與檢測；
在線服務計量與計費管理系統 搜索引擎的關鍵詞點擊計費；
此外，流式計算和增量計算也應用在工業4.0和物聯網上。
流式計算的數據特點
流（stream）是由業務產生的有向無界的數據流。
不可控性：你不知道數據的到達時機以及相關數據的順序，對於數據質量和規模也是不可控的；
時效性要求：在容錯方案、體系架構和結構輸出方面都與傳統的計算是截然不同的；
體系缺失：傳統學術領域已經對批量計算和離線計算的體系研究的非常成熟，而在實時領域如數據倉庫中間層等領域都是缺失的，包括數據源管理、數據質量管理等等。
另外，數據處理粒度最小，可以小到幾條數據，對架構產生決定性影響；
處理算子對全局狀態影響不同，有狀態、無狀態、順序不同等；
輸出要求，比如一致性和連貫性等。
整個流計算會對系統有非常多的不一樣的要求，這就會導致整個系統有非常大的複雜性，跟離線非常的不同，我們的計算仍然要求時效性、要求快，質量上要求它的計算一定是精準的，對容錯的要求，不論你的機器、集群、網絡硬件有任何的宕機，計算應該是持續穩定，對整個計算的要求也是非常多樣性的。關於多樣性，不同的業務場景，對計算的結果要求也是不一樣的，有些要求精確，一點數據都不能丟、精度損失，還有的業務場景要求可以多但是不能少，還有丟數據有一個sla在保證等，所以種種特點導致我們做流式計算和增量計算系統會面臨與傳統的離線計算和增量計算完全不同的要求。
與批量計算的區別
從架構角度，增量計算、流式計算和離線處理、批處理有什麼本質的區別?
與批量計算的區別如上圖所示，比如全量計算設計理念是面向吞吐，而流式計算是實時計算的一部分，面向延時；隨之而來的整個全量DAG是一個串型的DAG，是一個StageByStage的DAG，而流式計算的DAG是一個並行DAG，也就是說Batch跟Batch之間是完全可以並行的，離線的批量系統它的串型化和Streaming場景下的並行化，它們在整個數據的時效性上
有非常大的區別，特別是在Latency的體現。
典型系統計算概要分析
下面將向大家介紹業界比較經典的幾個流計算產品：
Twitter Storm
Storm是Twitter內部使用開源被廣泛使用的一套流計算系統，那麼它的一個核心概念是說，一個任務要創建一個Topology，它表示了一個完整的流計算作業，它的最開始的源頭名字叫做Spout，做收集數據的任務，它的前面可以掛任何的數據源、任何一個隊列系統甚至可以對接文件，那麼Bolt是它的具體計算任務所在的載體，而Bolt裡有諸多的Task，它是在Spout和Bolt裡負責具體一個數據分片的實體，它也是Storm裡調度的最小單位。Acker負責跟蹤消息是否被處理的節點。Storm的整個容錯是採用源頭重發的消息機制
源頭重發在網絡流量激增的情況下，會造成系統的雪崩風險大大提升。上圖是兩個Storm的作業，它先從源頭讀出數據，然後進行filter過濾，最終進行join，join後進行一些邏輯處理。
Nimbus–Zookeeper–Supervisor
Storm採用了Nimbus Supervisor之間的方式進行任務調度和跟蹤，它們之間是利用Zookeeper來進行通訊，Nimbus相當於一個全局的任務Master，負責接收Topology，然後進行二重的資源調度，並且將調度的信息記錄到Zookeeper中，定期檢查Zookeeper中的各種Supervisor的心跳信息，根據心跳狀態決定任務是否進行重新調度，而Supervsor充當著每臺物理機的一個watchdog，它在輪詢Zookeeper中的調度任務信息，然後接收到發現有啟動任務的信息，就會拉啟進程，啟動Task，同時定期要把心跳信息寫入Zookeeper，以便Supervisor來做出重新調度或者系統的重發操作。
消息跟蹤機制是Storm的核心，保證消息至少被處理一次，它追蹤源頭信息的所有子孫信息。
基本思路如下：
Acker節點是進行消息跟蹤的節點，以源頭消息的ID為hash key，來確定跟蹤的Acker，源頭信息對應的所有的子孫消息都有該Acker負責跟蹤，而消息樹上每產生一個新的子孫消息，則通知對應的Acker，子孫消息被處理，然後再去通知對應的Acker，當Acker裡所有的子孫消息都被處理的時候，那麼整個數據處理就完成了。
子孫的產生是由父節點，而處理是被子節點。所以Storm用了一個非常巧妙的異或方法，當父節點產生這個消息時，產生一個隨機數，把這個隨機數異或到Acker裡，Acker把這個隨機數傳遞到下一步的節點，當這個節點正確被處理以後，再把這節點發送給Acker去做異或，所以Storm利用了這個Acker機制來壓縮整個數據的跟蹤機制，最終保證任意節點出現宕機而值不會變成0。
Transactional Topology
光有以上的機制，還遠遠不夠。被系統重發的消息沒有任何附加信息，用戶無法判斷消息是否是被重發的等一些問題還有待解決，為解決消息被重複處理的問題，Storm 0.7.0以後版本推出了Transactional Topology進行改進，
原理如下：
在Spout上將源頭消息串行劃分成 Batch，為每個Batch賦以遞增的id，記錄在Zookeeper中，利用Acker跟蹤Batch是否被完全處理完成，超時或者節點異常，Spout重發Batch內的所有消息，不影響中間狀態的操作可以併發的執行，例如 Batch內的聚合操作，用戶代碼利用唯一的Batch ID進行去重。
整個Topology同一時刻只能有一個Batch正在提交，以保證在每個節點上Batch串行遞增，簡化用戶去重的邏輯。
Storm優缺點
優點：消息在框架內不落地，處理非常高效，保證了消息至少被處理，Transactional Topology為消息去重提供了可能，調度模式簡單，擴展能力強（關閉重發模式下），社區資源豐富，擁有各種常見消息源的Spout實現。
當然Storm也有自己的劣勢：Transactional Topology對Batch串行執行方式，性能下降嚴重；Batch太大太小都有問題，大小需要用戶根據具體業務分情況設置等。
Amazon Kinesis
Kinesis系統是一種完全託管的實時處理大規模數據流的開放服務。
所有節點運行於EC2中：相對Storm來說，它採用了消息節點內部重放的系統，而不是像Storm那樣子源頭重發，它的所有的節點都已經在EC2中，無需單獨的調度策略、複用安全、資源隔離機制，且擴展性好、彈性可伸縮。
只支持單級Task，可以利用多個Stream組成複雜的DAG-Task，用戶代碼需要實現DAG-Task內部的消息去重邏輯。
數據收集與計算獨立：數據收集模塊（Shard）對消息進行持久化，最長保留24小時；可以Get方式從其它系統中讀取Shard數據，計算模塊（Kinesis App）處理被推送的數據，Instance個數與Shard個數相同；用戶代碼可以自主控制Checkpoint節奏。
用戶可以自主調用相應的SplitShard\MergeShard接口，Stream上所有App的併發度隨之調整。具體實現方法如下：
每個Shard串行將接收到的消息寫入S3文件中，SplitShard後，原有Shard不再接收新數據，原有Shard對應的所有App的Instance處理完消息後關閉，啟動新的Shards(兩個)和對應新的Instances。
使計算可以更加的彈性，服務的可用性也更高。
Google MillWheel
MillWheel系統是利用內部支持Snapshot功能的Bigtable來進行持續化中間結果，將每個節點的計算輸出消息進行持久化，實現消息的“不丟不重”。
區別於Storm的是，它沒有複雜的跟蹤樹。因為每一級都把它的輸出消息進行持久化，用戶可以通過SetTimer\ProcessTimer接口解決用戶代碼在消息到來時才能取得控制流的弊端，然後在源頭節點（Injector）上將數據打上系統時間戳，每個內部節點（Computation）計算出所有輸入Pipe上的最小時間戳，向所有輸出Pipe上廣播當前完成的最小時間戳，用戶可以利用Low Watermark這一機制解決消息亂序或一致性問題。
核心技術
那麼，流式計算和增量計算中最核心的一些技術和難點有哪些呢？
從這張圖可以看到，整個流計算是由一個複雜的Topology所構成。那麼，從輸入到輸出，其中比較重要的兩個角色一是Jobmaster，一是Coordinator。Jobmaster是每個Job負責運行時的一個master；而Coordinator是剛才所說的消息跟蹤的一個角色，所以Coordinator最好是完全可以做到無狀態的線性擴展。
Batch數據從源頭進入後，進入Source節點，Source節點會從消息源讀取數據，藍色的部分代表著Worker節點，藍色節點再向橙色節點進行數據傳輸的時候，遵循著Shuffle的方法，可以是哈希的方法，可以是廣播的方法，也可以是任何用戶自定義的方法，output節點會將輸出結果向在線系統輸出，或者向下一級MQ節點輸出，輸出的結果也是按照Batch去對齊。
系統邊界-數據收集/結果數據
拉：從消息隊列（Kafka）、存儲（Hbase、HDFS）等系統讀取數據，並且藉助這些源頭已經持久化的數據實現系統的故障恢復；涉及第三方服務系統授權。
推：需要實現Http處理模塊（Apache、Nginx等），更需要解決故障恢復問題。
訂閱：結果數據寫入消息隊列，業務方訂閱，進入自己在線存儲系統。
服務：直接提供在線數據服務；涉及第三方服務系統授權；結果數據時鐘對齊。
shuffle機制
數據如何在處理節點之間流轉，這就涉及到shuffle機制了。
在流計算的處置機制下，數據採用了push模式，它整個數據不落地，對於數據傳輸的時效性是比較好的；另外，消息機制是需要解決丟數據和重數據的問題，框架也需要保證消息的有序性。
計算節點
流計算的計算進程是longlive的，即便沒有數據進程也在，所以決定了我們的系統不同的調度方式、不同的消息機制。計算的容錯採用任務跟蹤機制，最重要的一點，流計算的計算是個有狀態的計算，這個中間狀態的存儲方式、容錯關乎著整個計算的時效性、正確性、吞吐等。
分佈式挑戰和服務化訴求
離線計算系統對整個編程模型進行了約束，所以它在計算規模以及容錯上面已經有了非常成熟的研究，但是在流計算情況下，它在擴展能力上集群規模的上限是多少？計算作業是否可以線性增加？
用戶針可不可以重新定義等價的DAG來避免數據傾斜（犧牲性能），同時，如何避免傾斜帶來超時/雪崩的問題，數據動態的變化如何做到實時調整？
數據如何高可靠存儲，集群擴容、系統代碼升級時是否需要停止服務？單節點故障是否會導致整體服務的不可用？此外，我們不可避免地面臨著多租戶管理的問題，也就是授權鑑權 、資源隔離；還有計量計費、安全體系和運維體系怎麼保證等等。這都是做流計算系統面臨的一些核心問題。
增量計算語義
我們提出了新的模型MRM，它分為三個主要階段：
Local階段，是指Batch的Local操作，這個語意完全等同於MapReduce，可以認為離線計算或者全量計算，是一個只有一個Batch的增量計算，所以這時候的Map語意是完全一樣的；
Reduce階段是一個Aggregate階段，相當於在Batch內的數據做一次重新的組合，但是增量的語意告訴我們一定要做跨Batch的數據的結果合併。
Merge階段，跨批數據做全局的聚合計算。
上圖為一個簡單的wordcount例子，按照這樣的Batch去分，可以看到Map階段每個Batch的輸入、輸出，然後Reduce輸出，第一個Batch輸出7，第二個Batch輸入5，依此類推，初始化情況下OldValue為0，那麼，Merge很簡單，就進行0和7的合併操作。在這個case下面，進行一個count操作，就是7+0在Batch2，它把本次的Reduce結果5和上一次Merge的全局結果再進行一次Merge12，依此類推，所以大家可以看到對角線的顏色，這就是一輪的Merge的結果會成為下一輪的OldValue。
可以看到，貫穿始終的是一個很重要的Batch概念，它是系統跟蹤數據/時效性處理的最小單位。其實Batch是一個可以scale的概念，它可以退化為全量計算，也可以把一條數據放在一個Batch裡，可以做到時效性非常高。這兩者系統的吞吐、時效性會有截然不同的表現。
增量計算具備三要素：確定性、可加性和可逆性。
圖中case是說，假設在雙十一有兩行sql，首先按照賣家ID去Groupby，我要統計賣家的實時成交情況，是按照sellerid為11賣家，假設剛開始沒賣，第二行按照每10塊錢分制方圖，
依此類推，就像大家看到雙十一的阿里巴巴大屏一樣，它是實時滾動的。
sellerid為11的這個賣家剛開始沒賣東西，第0檔現在沒賣家，第一檔當前時刻已經有10個賣家了，第七檔有53位賣家，賣家為11做成了一個5塊錢的生意，那麼整個計算11這個0會變成5，現在屬於第一檔，那麼就把第一檔值由0變成1，但是緊接著，11號賣家又做成了16塊錢的生意，someMoney就會變成21，21屬於20到29區間，也就是這個檔位原來有10個賣家，現在變成11個賣家，但是重要的是，應該還把原來第0檔的那個1減掉，第一要把現在檔位加上去，還要把原來對其他檔位的貢獻要撤銷掉，這就是我們說的可逆性。
我們需要把這樣的UDEF，不僅要做一個正像操作，同時也要去實現一個負向操作。意思就是，把原來的那個值對系統狀態的影響讓用戶有機會撤銷掉，可以看到整個增量計算的一個
增量語意的rollback，相當於對數據庫領域的物化視圖，兩個物化視圖實時更新，一個物化視圖嵌套一個物化視圖，必須要對之前的計算有一個回穩操作，這是增量計算的一個非常本質的要求。
消息機制
消息機制是整個流處理系統的核心，它會影響計算延遲、容錯機制、可擴展性、服務可用性等方面，整個消息過程從分發到接收到處理相應用戶的work代碼，我們將這個消息框架定義為shuffle framework，對於消息的發送和接收利用Zookeeper來記錄整個DAG。也可以利用消息隊列比如Kafka，就像我們每一步都落地、存儲這樣的方式來定義shuffle。
對於解決消息的丟失問題，有兩種做法：
一是消息源頭的重發。每一個節點跟節點之間是用Push過去的，中間結果不會進行任何的可靠存儲，所以運行效率是比較高的，但是當一個Topology比較大的時候，任意一個節點的宕機或者超時都會引起整體的重發和重做，進而增加雪崩的風險；
一是節點內部重發。它跟前者的優勢劣勢剛好反轉，它需要每一步落地，運行效率會比較低，系統也會產生大量的文件碎片，它的好處是重發重做只依賴於父親節點。
消息源頭重發機制
方案
依賴源頭可靠的數據存儲系統（Kafka、HBase等），源頭存儲系統既是消息收集模塊，又是消息重發的數據源，中間節點消息不落地，跟蹤源頭消息，超時後重發。
雪崩
分佈式系統中最常見的異常狀態是網絡的抖動。在流處理系統中，消息跟蹤代價過大，一般的跟蹤機制並不會在跟蹤結果中詳細標示出是哪一個節點出現故障。這種不加區分的源頭消息重發，會使得本來正常的節點因為其它單點的故障，也要接收大量的重複數據，消耗寶貴的網絡資源，使網絡狀況進一步惡化。然後這種情況會一步一步地變大，最終造成整個集群網絡的癱瘓。
針對以上問題，我們進行了消息命令流的大量約減，我們不會跟蹤具體每一個Batch內的數據，而是每一批發送一個特殊的命令流來跟蹤。我們要求每個消息有唯一的batchid，並且與源頭節點的offset可重入，還需要這樣的Batchid去做消重，也作為一個唯一的版本，每次重試我們會在Batchid後面新增一個attemptid。
當然這樣的問題還是存在大量的通訊量，節點Crash後整個表如何重建等問題。
消息節點內部重放
方案
節點接收到消息後，先落地，再計算；節點出現故障，從存儲系統中重放；定期做 Checkpoint，減少重放代價。
這種方案特點是方案簡單，但依賴於可靠、高效的存儲模塊；局部故障對全局影響小，系統可擴展性好，雪崩的風險也極大地降低。
消息去重
消息去重策略是上下游節點之間通過自增的ID協議。
發送端消息從0開始賦值唯一的id，每次加1；發送消息後等待接收端返回信息，成功或者消息重複才可發送下一消息，否則一直重試。
接收端在內存維護一個去重表，Key是上游節點對應的ID，Value是對應上游節點最後一次接收成功的消息id；接到新消息，首先在去重表中，根據PipeID比較消息id是否自增加1；如果小於等於已有消息id，則直接返回消息重複；否則，將消息寫入存儲系統中，之後更新去重表，並返回發送端消息接收成功。
為了避免造成大量的IO浪費，我們也會使用讀寫分離技術。
寫模塊：
節點將接收到的上游消息首先進行持久化；
將數據和存儲系統中的Offset信息通過內存（或者網絡）嘗試發送至處理模塊的緩衝區中；
如果緩衝區不滿，直接放置隊尾；
如果緩衝區已滿，則將緩衝區中隊首尚未處理的消息丟棄，然後將當前消息放置隊尾。
讀模塊：
消息處理模塊讀取緩衝區隊首的消息；
通過判斷該消息的Offset信息是否連續來確定是否有消息丟失；
如果發現消息丟失，直接從存儲系統“追讀”丟失的消息，直到追上緩衝區隊首的消息。
讀寫分離的好處是，網絡抖動不影響其它節點，局部故障也不放大，不會出現處理快的節點一直在等慢的節點等。
有狀態計算
增量計算
Map-Reduce-Merge模型，Map和Reduce約定在Batch內的一個Map操作和aggregate的一個reduce操作，語意和Map-Reduce完全兼容；用戶只需要在Merge裡面去寫用戶邏輯，全部是由增量計算框架來維持，輸入這個Value是本次Reduce的結果，用戶只用把oldValue和這個value進行合併操作並且返回新的值，作為下一次的oldValue傳。
Checkpoint
當作業的併發數增長時，一定程度上，任務的併發程度的增長已經受限於Hbase的能力，尤其是隨機讀的能力，所以對整個系統的scale ability是有限制的。
對此，我們引入了一個內存增量的snapshot機制。用戶可以指定固定批次的數據，在這個批次內的對於OldValue和state的修改，完全都在一個增量的snapshot，而這個snapshot在內存裡頭， Merge是update，它完全是在增量的snapshot內完成，而系統這時持續地引進會產生大量增量的snapshot，這時系統會在內部啟動一個checkpoint的線程，它會順序地將這些snapshot選擇性地進行Compact，Compact後將這個內存的snapshot持久化，批量地刷入到盤古，刷入到一個全局的快存儲。
這樣的機制既保證用戶在調用Merge這個函數的時候，基本上都在操作內存，而整個系統的scale ability不依賴於其他第三方的可靠存儲，而系統將snapshot的checkpoint是在異步的後臺進行。
並行DAG
為了克服datascale以及增加系統的時效性，整個DAG完全是一個並行的DAG。
那進行一個簡單的建模：假設有N條數據，M個資源，共有n個module，第i個module的吞吐為OI,調度的資源數為Pi。
圖中可以看出，在理想情況下，它們完成的延時是一致的。
但事實上完全不是。限制時下的物理模型遠遠比這個複雜，串行的模式的優勢是模型簡單、吞吐高。劣勢是數據時效性和數據傾斜對系統的整體延時傷害。所以串行模型是面向吞吐、兼顧延時。
並行DAG優勢是數據時效性好、對傾斜友好。但是它的建模非常複雜，調度也是非常複雜。
並行模型是面向延時而兼顧吞吐的。
搶佔式調度和資源隔離
整個流計算是個longlive的進程，所以業界之前的調度系統針對任務結束後進程回收的情況很明顯不再適用，那麼離線裡面，無論是Yarn還是fuxi，都不能適應長進程的任務調度。
現在有一個開源項目叫Slider，它在某種程度上嘗試去解決這個問題。在線系統的調度與離線系統的調度差異性是非常大的，我們需要解決的問題還有很多很多。
在隔離維度上，用戶程序使用的Memory、Network、CPU（隔離難度依次遞增），不需要隔離本地IO訪問（不容許用戶程序訪問本地IO），框架使用的資源，通過消息“流控”來限制。
而Memory上，Java程序通過啟動jvm時的-Xss參數設置；C\C++程序通過定期查看linux下/proc/pid/status，超過上限後，限制消息輸入；Network通過linux系統的iptables + tc模塊。
CPU通過linux taskset命令將進程綁定到某一具體CPU上，同一CPU上多進程依然會發生搶佔資源。
Failover機制
整個流計算的Failover容錯機制，Batch是容錯的最小單位，是數據跟蹤的最小單位，是輸入輸出的最小單位，是控制的最小單位；整個容錯分為源頭重建和節點重建兩種，全量輸出，無外部互相依賴，跟蹤消息與消息體量級。
離線跟蹤、流式跟蹤、在線跟蹤，完全在實現方法上、策略上不一樣。
那麼有狀態計算的Failover的checkpoint，它的內存重建，大家可以關注開源的tachyon，在整個Failover的機制設計方面，有運行時效率和恢復時效率的一個tradeoff，包括如何避免雪崩，這些都是在容錯機制上要考慮的重點問題。
綜上所述，整個系統是在不斷做TradeOff：
吞吐與響應時間的TradeOff
實時性與數據鏈路的不可控的TradeOff
非冪等操作與數據鏈路的不可控的TradeOff
精度與成本的TradeOff
恢復成本與運行時成本的TradeOff
全鏈路與系統邊界的Tradeoff
需求多樣性與平臺一致性的TradeOff
不同計算場景不同技術體系的TradeOff
StreamSQL
streamSQL支持用SQL一樣的語意來表示，讓用戶使用SQL來操作流計算。streamSQL提供了5個基本原語：Map、Reduce 、Shuffle 、Union和Merge。我們也實現了一些高級算子，用戶可以去擴展高級算子，包括Topk、distinct 、Join 、windows。
如圖所示，最底下的引擎是整個流計算的一個並行DAG；MRM層最主要的部分分為消息控制、容錯以及計算模型；在此之上定義了算子層；算子層再往上是SQL層，SQL層分為SQL解析、制定邏輯執行計劃、根據拓普運行情況和原信息進行物理執行計劃的優化和SQL的改寫。
那麼我們可以定義我們的源表，也就是數據源，你可以create一個stream table。用戶還可以定義自己的維表、臨時表，臨時表作為一個SQL的極聯存在，它定義了內部數據流的一個Schema，實際上它是不存儲任何的數據，只是做串聯上下游的邏輯，構造複雜DAG。
用戶除了寫SQL之外，還可以自定義函數算子來實現它的邏輯。
StreamSQL的系統架構如圖所示，分成gateway層、控制集群層和worker執行層。
用戶看到的是統一的邏輯集群，我們提供了開發平臺、Web UI、監控報警以及profiling 和Metric子系統等。
延伸
在實際業務場景中，會發現很多很多的問題等待我們去解決，比如：
無法做任務之間的複用數據
完成業務需要各種計算模型
多個系統融合
系統之間無法共享數據
離線與在線的鴻溝依然明顯
以上就是跟大家分享的流式計算和增量計算的關鍵技術點，謝謝大家！