幾個月之前我們在這里討論過[](http://www.cakesolutions.net/teamblogs/introduction-into-distributed-real-time-stream-processing)目前對于這種日漸增加的分布式流計算的需求的原因。當然，目前也有很多的各式各樣的框架被用于處理這一些問題。現在我們會在這篇文章中進行回顧，來討論下各種框架之間的相似點以及區別在哪里，還有就是從我的角度分析的，推薦的適用的用戶場景。

如你所想，分布式的流處理也就是通常意義上的持續處理、數據富集以及對于無界數據的分析過程的組合。它是一個類似于MapReduce這樣的通用計算模型，但是我們希望它能夠在毫秒級別或者秒級別完成響應。這些系統經常被有向非循環圖(Directed ACyclic Graphs,DAGs)來表示。

DAG主要功能即是用圖來表示鏈式的任務組合，而在流處理系統中，我們便常常用DAG來描述一個流工作的拓撲。筆者自己是從Akka的Stream 中的術語得到了啟發。如下圖所示，數據流經過一系列的處理器從源點流動到了終點，也就是用來描述這流工作。談到Akka的Streams，我覺得要著重強調下分布式這個概念，因為即使也有一些單機的解決方案可以創建并且運行DAG，但是我們仍然著眼于那些可以運行在多機上的解決方案。

Points of Interest

在不同的系統之間進行選擇的時候，我們主要關注到以下幾點。

• Runtime and Programming model(運行與編程模型)

一個平臺提供的編程模型往往會決定很多它的特性，并且這個編程模型應該足夠處理所有可能的用戶案例。這是一個決定性的因素，我也會在下文中多次討論。

• Functional Primitives(函數式單元)

一個合格的處理平臺應該能夠提供豐富的能夠在獨立信息級別進行處理的函數，像map、filter這樣易于實現與擴展的一些函數。同樣也應提供像aggregation這樣的跨信息處理函數以及像join這樣的跨流進行操作的函數，雖然這樣的操作會難以擴展。

• State Management(狀態管理)

大部分這些應用都有狀態性的邏輯處理過程，因此，框架本身應該允許開發者去維護、訪問以及更新這些狀態信息。

• Message Delivery Guarantees(消息投遞的可達性保證)

一般來說，對于消息投遞而言，我們有至多一次(at most once)、至少一次(at least once)以及恰好一次(exactly once)這三種方案。

• at most once

At most once投遞保證每個消息會被投遞0次或者1次，在這種機制下消息很有可能會丟失。

• at least once

At least once投遞保證了每個消息會被默認投遞多次，至少保證有一次被成功接收，信息可能有重復，但是不會丟失。

• exactly once

exactly once意味著每個消息對于接收者而言正好被接收一次，保證即不會丟失也不會重復。

• Failures Handling

在一個流處理系統中，錯誤可能經常在不同的層級發生，譬如網絡分割、磁盤錯誤或者某個節點莫名其妙掛掉了。平臺要能夠從這些故障中順利恢復，并且能夠從最后一個正常的狀態繼續處理而不會損害結果。

除此之外，我們也應該考慮到平臺的生態系統、社區的完備程度，以及是否易于開發或者是否易于運維等等。

RunTime and Programming Model

運行環境與編程模型可能是某個系統的最重要的特性，因為它定義了整個系統的呈現特性、可能支持的操作以及未來的一些限制等等。因此，運行環境與編程模型就確定了系統的能力與適用的用戶案例。目前，主要有兩種不同的方法來構建流處理系統,其中一個叫Native Streaming，意味著所有輸入的記錄或者事件都會根據它們進入的順序一個接著一個的處理。

 

另一種方法叫做Micro-Batching。大量短的Batches會從輸入的記錄中創建出然后經過整個系統的處理，這些Batches會根據預設好的時間常量進行創建，通常是每隔幾秒創建一批。

兩種方法都有一些內在的優勢與不足，首先來談談Native Streaming。好的一方面呢是Native Streaming的表現性會更好一點，因為它是直接處理輸入的流本身的，并沒有被一些不自然的抽象方法所限制住。同時，因為所有的記錄都是在輸入之后立馬被處理，這樣對于請求方而言響應的延遲就會優于那種Micro-Batching系統。處理這些，有狀態的操作符也會更容易被實現，我們在下文中也會描述這個特點。不過Native Streaming系統往往吞吐量會比較低，并且因為它需要去持久化或者重放幾乎每一條請求，它的容錯的代價也會更高一些。并且負載均衡也是一個不可忽視的問題，舉例而言，我們根據鍵對數據進行了分割并且想做進一步地處理。如果某些鍵對應的分區因為某些原因需要更多地資源去處理，那么這個分區往往就會變成整個系統的瓶頸。

而對于Micro-Batching而言，將流切分為小的Batches不可避免地會降低整個系統的變現性，也就是可讀性。而一些類似于狀態管理的或者joins、splits這些操作也會更加難以實現，因為系統必須去處理整個Batch。另外，每個Batch本身也將架構屬性與邏輯這兩個本來不應該被糅合在一起的部分相連接了起來。而Micro-Batching的優勢在于它的容錯與負載均衡會更加易于實現，它只要簡單地在某個節點上處理失敗之后轉發給另一個節點即可。最后，值得一提的是，我們可以在Native Streaming的基礎上快速地構建Micro-Batching的系統。

而對于編程模型而言，又可以分為Compositional(組合式)與Declarative(聲明式)。組合式會提供一系列的基礎構件，類似于源讀取與操作符等等，開發人員需要將這些基礎構件組合在一起然后形成一個期望的拓撲結構。新的構件往往可以通過繼承與實現某個接口來創建。另一方面，聲明式API中的操作符往往會被定義為高階函數。聲明式編程模型允許我們利用抽象類型和所有其他的精選的材料來編寫函數式的代碼以及優化整個拓撲圖。同時，聲明式API也提供了一些開箱即用的高等級的類似于窗口管理、狀態管理這樣的操作符。下文中我們也會提供一些代碼示例。

Apache Streaming Landscape

目前已經有了各種各樣的流處理框架，自然也無法在本文中全部攘括。所以我必須將討論限定在某些范圍內，本文中是選擇了所有Apache旗下的流處理的框架進行討論，并且這些框架都已經提供了Scala的語法接口。主要的話就是Storm以及它的一個改進Trident Storm，還有就是當下正火的Spark。最后還會討論下來自LinkedIn的Samza以及比較有希望的Apache Flink。筆者個人覺得這是一個非常不錯的選擇，因為雖然這些框架都是出于流處理的范疇，但是他們的實現手段千差萬別。

Apache Storm 最初由Nathan Marz以及他的BackType的團隊在2010年創建。后來它被Twitter收購并且開源出來，并且在2014年變成了Apache的頂層項目。毫無疑問，Storm是大規模流處理中的先行者并且逐漸成為了行業標準。Storm是一個典型的Native Streaming系統并且提供了大量底層的操作接口。另外，Storm使用了Thrift來進行拓撲的定義，并且提供了大量其他語言的接口。

Trident 是一個基于Storm構建的上層的Micro-Batching系統，它簡化了Storm的拓撲構建過程并且提供了類似于窗口、聚合以及狀態管理等等沒有被Storm原生支持的功能。另外，Storm是實現了至多一次的投遞原則，而Trident實現了恰巧一次的投遞原則。Trident 提供了 Java, Clojure 以及 Scala 接口。

眾所周知，Spark是一個非常流行的提供了類似于SparkSQL、Mlib這樣內建的批處理框架的庫，并且它也提供了 Spark Streaming這樣優秀地流處理框架。Spark的運行環境提供了批處理功能，因此，Spark Streaming毫無疑問是實現了Micro-Batching機制。輸入的數據流會被接收者分割創建為Micro-Batches，然后像其他 Spark任務一樣進行處理。Spark 提供了 Java, Python 以及 Scala 接口。

Samza最早是由LinkedIn提出的與Kafka協同工作的優秀地流解決方案，Samza已經是LinkedIn內部關鍵的基礎設施之一。Samza重負依賴于Kafaka的基于日志的機制，二者結合地非常好。Samza提供了Compositional接口，并且也支持Scala。

最后聊聊Flink. Flink可謂一個非常老的項目了，最早在2008年就啟動了，不過目前正在吸引越來越多的關注。Flink也是一個Native Streaming的系統，并且提供了大量高級別的API。Flink也像Spark一樣提供了批處理的功能，可以作為流處理的一個特殊案例來看。 Flink強調萬物皆流，這是一個絕對的更好地抽象，畢竟確實是這樣。

下表就簡單列舉了上述幾個框架之間的特性：

 

Counting Words

Wordcount就好比流處理領域的HelloWorld，它能夠很好地描述不同框架間的差異性。首先看看Storm是如何編寫WordCount程序的：

TopologyBuilder builder = new TopologyBuilder();
 builder.setSpout("spout", new RandomSentenceSpout(), 5);
 builder.setBolt("split", new Split(), 8).shuffleGrouping("spout");
 builder.setBolt("count", new WordCount(), 12).fieldsGrouping("split", new Fields("word"));

 ...

 Map

首先來看看它的拓撲定義，在第2行那邊是定義了一個Spout，也就是一個輸入源。然后定義了一個Bold，也就是一個處理的組件，用于將某個句子分割成詞序列。然后還定義了另一個Bolt用來負責真實的詞計算。5，8到12行省略的過程用于定義集群中使用了多少個線程來供每一個組件使用。如你所見，所有的定義都是比較底層的與手動的。接下來繼續看看這個8-15行，也就是真正用于WordCount的部分代碼。因為Storm沒有內建的狀態處理的支持，所以我必須自定義這樣一個本地狀態，和理想的相差甚遠啊。下面我們繼續看看Trident。

正如我上文中提及的，Trident是一個基于Storm的Micro-Batching的擴展，它提供了狀態管理等等功能。

public static StormTopology buildTopology(LocalDRPC drpc) {
 FixedBatchSpout spout = ...

 TridentTopology topology = new TridentTopology();
 TridentState wordCounts = topology.newStream("spout1", spout)
 .each(new Fields("sentence"),new Split(), new Fields("word"))
 .groupBy(new Fields("word"))
 .persistentAggregate(new MemoryMapState.Factory(), 
 new Count(), new Fields("count"));

 ...

 }

從代碼中就可以看出，在Trident中就可以使用一些上層的譬如each、groupBy這樣的操作符，并且可以在Trident中內建的進行狀態管理了。接下來我們再看看Spark提供的聲明式的接口，要記住，與前幾個例子不同的是，基于Spark的代碼已經相當簡化了，下面基本上就是要用到的全部的代碼了：

val conf = new SparkConf().setAppName("wordcount")
val ssc = new StreamingContext(conf, Seconds(1))

val text = ...

val counts = text.flatMap(line => line.split(" "))
 .map(word => (word, 1))
 .reduceByKey(_ + _)

counts.print()

ssc.start()
ssc.awaitTermination()

每個Spark的流任務都需要一個StreamingContext用來指定整個流處理的入口。StreamingContext定義了Batch的間隔，上面是設置到了1秒。在6-8行即是全部的詞統計的計算過程，非常不一樣啊。下面再看看Apache Samza，另一個代表性的組合式的API：

class WordCountTask extends StreamTask {

  override def process(envelope: IncomingMessageEnvelope, collector: MessageCollector, 
    coordinator: TaskCoordinator) {

    val text = envelope.getMessage.asInstanceOf[String]

    val counts = text.split(" ").foldLeft(Map.empty[String, Int]) {
      (count, word) => count + (word -> (count.getOrElse(word, 0) + 1))
    }

    collector.send(new OutgoingMessageEnvelope(new SystemStream("kafka", "wordcount"), counts))

 }

Topology定義在了Samza的屬性配置文件里，為了明晰起見，這里沒有列出來。下面再看看Fink，可以看出它的接口風格非常類似于Spark Streaming，不過我們沒有設置時間間隔：

val env = ExecutionEnvironment.getExecutionEnvironment

 val text = env.fromElements(...)
 val counts = text.flatMap ( _.split(" ") )
   .map ( (_, 1) )
   .groupBy(0)
   .sum(1)

 counts.print()

 env.execute("wordcount")

與批處理系統相比，流處理系統中的容錯機制固然的會比批處理中的要難一點。在批處理系統中，如果碰到了什么錯誤，只要將計算中與該部分錯誤關聯的重新啟動就好了。不過在流計算的場景下，容錯處理會更加困難，因為會不斷地有數據進來，并且有些任務可能需要7*24地運行著。另一個我們碰到的挑戰就是如何保證狀態的一致性，在每天結束的時候我們會開始事件重放，當然不可能所有的狀態操作都會保證冪等性。下面我們就看看其他的系統是怎么處理的：

Storm

Storm使用了所謂的逆流備份與記錄確認的機制來保證消息會在某個錯誤之后被重新處理。記錄確認這一個操作工作如下：一個操作器會在處理完成一個記錄之后向它的上游發送一個確認消息。而一個拓撲的源會保存有所有其創建好的記錄的備份。一旦受到了從Sinks發來的包含有所有記錄的確認消息，就會把這些確認消息安全地刪除掉。當發生錯誤時，如果還沒有接收到全部的確認消息，就會從拓撲的源開始重放這些記錄。這就確保了沒有數據丟失，不過會導致重復的 Records處理過程，這就屬于At-Least投送原則。

Storm用一套非常巧妙的機制來保證了只用很少的字節就能保存并且追蹤確認消息，但是并沒有太多關注于這套機制的性能，從而使得Storm有較低地吞吐量，并且在流控制上存在一些問題，譬如這種確認機制往往在存在背壓的時候錯誤地認為發生了故障。

 

Spark Streaming

Spark Streaming以及它的Micro-Batching機制則使用了另一套方案，道理很簡單，Spark將Micro-Batches分配到多個節點運行，每個Micro-Batch可以成功運行或者發生故障，當發生故障時，那個對應的Micro-Batch只要簡單地重新計算即可，因為它是持久化并且無狀態的，所以要保證Exactly-Once這種投遞方式也是很簡單的。

Samza

Samza的實現手段又不一樣了，它利用了一套可靠地、基于Offset的消息系統，在很多情況下指的就是Kafka。Samza會監控每個任務的偏移量，然后在接收到消息的時候修正這些偏移量。Offset可以是存儲在持久化介質中的一個檢查點，然后在發生故障時可以進行恢復。不過問題在于你并不知道恢復到上一個CheckPoint之后到底哪個消息是處理過的，有時候會導致某些消息多次處理，這也是At-Least的投遞原則。

Flink

Flink主要是基于分布式快照，每個快照會保存流任務的狀態。鏈路中運送著大量的CheckPoint Barrier(檢查點障礙，就是分隔符、標識器之類的)，當這些Barrier到達某個Operator的時候，Operator將自身的檢查點與流相關聯。與Storm相比，這種方式會更加高效，畢竟不用對每個Record進行確認操作。不過要注意的是，Flink還是Native Streaming，概念上和Spark還是相去甚遠的。Flink也是達成了Exactly-Once投遞原則。

Managing State

大部分重要的流處理應用都會保有狀態，與無狀態的操作符相比，這些應用中需要一個輸入和一個狀態變量，然后進行處理最終輸出一個改變了的狀態。我們需要去管理、存儲這些狀態，要保證在發生故障的時候能夠重現這些狀態。狀態的重造可能會比較困難，畢竟上面提到的不少框架都不能保證Exactly- Once，有些Record可能被重放多次。

Storm

Storm是實踐了At-Least投遞原則，而怎么利用Trident來保證Exactly-Once呢?概念上還是很簡單的，只需要使用事務進行提交Records，不過很明顯這種方式及其低效。所以呢，還是可以構建一些小的Batches，并且進行一些優化。Trident是提供了一些抽象的接口來保證實現Exactly-Once，如下圖所示，還有很多東西等著你去挖掘。

Spark Streaming

當想要在流處理系統中實現有狀態的操作時，我們往往想到的是一個長時間運行的Operator，然后輸入一個狀態以及一系列的Records。不過 Spark Streaming是以另外一種方式進行處理的，Spark Streaming將狀態作為一個單獨地Micro-Batching流進行處理，所以在對每個小的Micro-Spark任務進行處理時會輸入一個當前的狀態和一個代表當前操作的函數，最后輸出一個經過處理的Micro-Batch以及一個更新好的狀態。

Samza

Samza的處理方式更加簡單明了，就是把它們放到Kafka中，然后問題就解決了。Samza提供了真正意義上的有狀態的Operators，這樣每個任務都能保有狀態，然后所有狀態的變化都會被提交到Kafka中。在有需要的情況下某個狀態可以很方便地從Kafka的Topic中完成重造。為了提高效率，Samza允許使用插件化的鍵值本地存儲來避免所有的消息全部提交到Kafka。這種思路如下圖所示，不過Samza只是提高了At- Least這種機制，未來可能會提供Exactly-Once。

Flink

Flink提供了類似于Samza的有狀態的Operator的概念，在Flink中，我們可以使用兩種不同的狀態。第一種是本地的或者叫做任務狀態，它是某個特定的Operator實例的當前狀態，并且這種狀態不會與其他進行交互。另一種呢就是維護了整個分區的狀態。

Counting Words with State

Trident

public static StormTopology buildTopology(LocalDRPC drpc) {
   FixedBatchSpout spout = ...

   TridentTopology topology = new TridentTopology();
  
   TridentState wordCounts = topology.newStream("spout1", spout)
     .each(new Fields("sentence"),new Split(), new Fields("word"))
     .groupBy(new Fields("word"))
     .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"));

 ...

 }

在第9行中，我們可以通過調用一個持久化的聚合函數來創建一個狀態。

Spark Streaming

// Initial RDD input to updateStateByKey
val initialRDD = ssc.sparkContext.parallelize(List.empty[(String, Int)])

val lines = ...
val words = lines.flatMap(_.split(" "))
val wordDstream = words.map(x => (x, 1))

val trackStateFunc = (batchTime: Time, word: String, one: Option[Int], 
  state: State[Int]) => {
    val sum = one.getOrElse(0) + state.getOption.getOrElse(0)
    val output = (word, sum)
    state.update(sum)
    Some(output)
  }

val stateDstream = wordDstream.trackStateByKey(
  StateSpec.function(trackStateFunc).initialState(initialRDD))

在第2行中，我們創建了一個RDD用來保存初始狀態。然后在5，6行中進行一些轉換，接下來可以看出，在8-14行中，我們定義了具體的轉換方程，即輸入時一個單詞、它的統計數量和它的當前狀態。函數用來計算、更新狀態以及返回結果，最后我們將所有的Bits一起聚合。

Samza

class WordCountTask extends StreamTask with InitableTask {

  private var store: CountStore = _

  def init(config: Config, context: TaskContext) {
    this.store = context.getStore("wordcount-store")
      .asInstanceOf[KeyValueStore[String, Integer]]
  }

 override def process(envelope: IncomingMessageEnvelope,
   collector: MessageCollector, coordinator: TaskCoordinator) {

   val words = envelope.getMessage.asInstanceOf[String].split(" ")

   words.foreach { key =>
     val count: Integer = Option(store.get(key)).getOrElse(0)
     store.put(key, count + 1)
     collector.send(new OutgoingMessageEnvelope(new SystemStream("kafka", "wordcount"), 
       (key, count)))
   }
 }

在上述代碼中第3行定義了全局的狀態，這里是使用了鍵值存儲方式，并且在5~6行中定義了如何初始化。然后，在整個計算過程中我們都使用了該狀態。

Flink

val env = ExecutionEnvironment.getExecutionEnvironment

val text = env.fromElements(...)
val words = text.flatMap ( _.split(" ") )

words.keyBy(x => x).mapWithState {
  (word, count: Option[Int]) =>
    {
      val newCount = count.getOrElse(0) + 1
      val output = (word, newCount)
      (output, Some(newCount))
    }
}

在第6行中使用了mapWithState函數，第一個參數是即將需要處理的單次，第二個參數是一個全局的狀態。

Performance

合理的性能比較也是本文的一個重要主題之一。不同的系統的解決方案差異很大，因此也是很難設置一個無偏的測試。通常而言，在一個流處理系統中，我們常說的性能就是指延遲與吞吐量。這取決于很多的變量，但是總體而言標準為如果單節點每秒能處理500K的Records就是個合格的，如果能達到100萬次以上就已經不錯了。每個節點一般就是指24核附帶上24或者48GB的內存。

對于延遲而言，如果是Micro-Batch的話往往希望能在秒級別處理。如果是Native Streaming的話，希望能有百倍的減少，調優之后的Storm可以很輕易達到幾十毫秒。

另一方面，消息的可達性保證、容錯以及狀態管理都是需要考慮進去的。譬如如果你開啟了容錯機制，那么會增加10%到15%的額外消耗。除此之外，以文章中兩個WordCount為例，第一個是無狀態的WordCount，第二個是有狀態的WordCount，后者在Flink中可能會有25%額外的消耗，而在Spark中可能有50%的額外消耗。當然，我們肯定可以通過調優來減少這種損耗，并且不同的系統都提供了很多的可調優的選項。

還有就是一定要記住，在分布式環境下進行大數據傳輸也是一件非常昂貴的消耗，因此我們要利用好數據本地化以及整個應用的序列化的調優。

Project Maturity(項目成熟度)

在為你的應用選擇一個合適的框架的時候，框架本身的成熟度與社區的完備度也是一個不可忽略的部分。Storm是第一個正式提出的流處理框架，它已經成為了業界的標準并且被應用到了像Twitter、Yahoo、Spotify等等很多公司的生產環境下。Spark則是目前最流行的Scala的庫之一，并且Spark正逐步被更多的人采納，它已經成功應用在了像Netflix、Cisco、DataStax、Indel、IBM等等很多公司內。而 Samza最早由LinkedIn提出，并且正在運行在幾十個公司內。Flink則是一個正在開發中的項目，不過我相信它發展的會非常迅速。

Summary

在我們進最后的框架推薦之前，我們再看一下上面那張圖：

 

Framework Recommendations

這個問題的回答呢，也很俗套，具體情況具體分析?？偟膩碚f，你首先呢要仔細評估下你應用的需求并且完全理解各個框架之間的優劣比較。同時我建議是使用一個提供了上層接口的框架，這樣會更加的開發友好，并且能夠更快地投入生產環境。不過別忘了，絕大部分流應用都是有狀態的，因此狀態管理也是不可忽略地一個部分。同時，我也是推薦那些遵循Exactly-Once原則的框架，這樣也會讓開發和維護更加簡單。不過不能教條主義，畢竟還是有很多應用會需要 At-Least-Once與At-Most-Once這些投遞模式的。最后，一定要保證你的系統可以在故障情況下很快恢復，可以使用Chaos Monkey或者其他類似的工具進行測試。在我們之前的討論中也發現這個快速恢復的能力至關重要。

對于小型與需要快速響應地項目，Storm依舊是一個非常好的選擇，特別是在你非常關注延遲度的情況下。不過還是要謹記容錯機制和 Trident的狀態管理會嚴重影響性能。Twitter目前正在設計新的流計算系統Heron用來替代Storm，它可以在單個項目中有很好地表現。不過Twitter可不一定會開源它。

對于Spark Streaming而言，如果你的系統的基礎架構中已經使用了Spark，那還是很推薦你試試的。另一方面，如果你想使用Lambda架構，那Spark 也是個不錯的選擇。不過你一定要記住，Micro-Batching本身的限制和延遲對于你而言不是一個關鍵因素。

如果你想用Samza的話，那最好Kafka已經是你的基礎設施的一員了。雖然在Samza中Kafka只是個可插拔的組件，不過基本上所有人都會使用Kafka。正如上文所說，Samza提供了強大的本地存儲功能，能夠輕松管理數十G的狀態數據。不過它的At-Least-Once的投遞限制也是很大一個瓶頸。

Flink目前在概念上是一個非常優秀的流處理系統，它能夠滿足大部分的用戶場景并且提供了很多先進的功能，譬如窗口管理或者時間控制。所以當你發現你需要的功能在Spark當中無法很好地實現的時候，你可以考慮下Flink。另外，Flink也提供了很好地通用的批處理的接口，只不過你需要很大的勇氣來將你的項目結合到Flink中，并且別忘了多關注關注它的路線圖。

Dataflow與開源

我最后一個要提到的就是Dataflow和它的開源計劃。Dataflow是Google云平臺的一個組成部分，是目前在Google內部提供了統一的用于批處理與流計算的服務接口。譬如用于批處理的MapReduce，用于編程模型定義的FlumeJava以及用于流計算的MillWheel。 Google最近打算開源這貨的SDK了，Spark與Flink都可以成為它的一個運行驅動。

Conclusion

本文我們過了一遍常用的流計算框架，它們的特性與優劣對比，希望能對你有用吧。