Stream 主流流处理框架比较(2)

在上篇文章中，我们过了下基本的理论，也介绍了主流的流处理框架：Storm，Trident，Spark Streaming，Samza和Flink。今天咱们来点有深度的主题，比如，容错，状态管理或者性能。除此之外，我们也将讨论开发分布式流处理应用的指南，并给出推荐的流处理框架。

1. 容错性

流处理系统的容错性与生俱来的比批处理系统难实现。当批处理系统中出现错误时，我们只需要把失败的部分简单重启即可；但对于流处理系统，出现错误就很难恢复。因为线上许多作业都是7 x 24小时运行，不断有输入的数据。流处理系统面临的另外一个挑战是状态一致性，因为重启后会出现重复数据，并且不是所有的状态操作是幂等的。容错性这么难实现，那下面我们看看各大主流流处理框架是如何处理这一问题。

1.1 Apache Storm

Storm使用上游数据备份和消息确认的机制来保障消息在失败之后会重新处理。消息确认原理：每个操作都会把前一次的操作处理消息的确认信息返回。Topology的数据源备份它生成的所有数据记录。当所有数据记录的处理确认信息收到，备份即会被安全拆除。失败后，如果不是所有的消息处理确认信息收到，那数据记录会被数据源数据替换。这保障了没有数据丢失，但数据结果会有重复，这就是at-least once传输机制。

Storm采用取巧的办法完成了容错性，对每个源数据记录仅仅要求几个字节存储空间来跟踪确认消息。纯数据记录消息确认架构，尽管性能不错，但不能保证exactly once消息传输机制，所有应用开发者需要处理重复数据。Storm存在低吞吐量和流控问题，因为消息确认机制在反压下经常误认为失败。

1.2 Spark Streaming

Spark Streaming实现微批处理，容错机制的实现跟Storm不一样。微批处理的想法相当简单。Spark在集群各worker节点上处理micro-batches。每个micro-batches一旦失败，重新计算就行。因为micro-batches本身的不可变性，并且每个micro-batches也会持久化，所以exactly once传输机制很容易实现。

1.3 Samza

Samza的实现方法跟前面两种流处理框架完全不一样。Samza利用消息系统Kafka的持久化和偏移量。Samza监控任务的偏移量，当任务处理完消息，相应的偏移量被移除。消息的偏移量会被checkpoint到持久化存储中，并在失败时恢复。但是问题在于：从上次checkpoint中修复偏移量时并不知道上游消息已经被处理过，这就会造成重复。这就是at least once传输机制。

1.4 Apache Flink

Flink的容错机制是基于分布式快照实现的，这些快照会保存流处理作业的状态(本文对Flink的检查点和快照不进行区分，因为两者实际是同一个事物的两种不同叫法。Flink构建这些快照的机制可以被描述成分布式数据流的轻量级异步快照，它采用Chandy-Lamport算法实现。)。如果发生失败的情况，系统可以从这些检查点进行恢复。Flink发送checkpoint的栅栏（barrier）到数据流中（栅栏是Flink的分布式快照机制中一个核心的元素），当checkpoint的栅栏到达其中一个operator，operator会接所有收输入流中对应的栅栏（比如，图中checkpoint n对应栅栏n到n-1的所有输入流，其仅仅是整个输入流的一部分）。所以相对于Storm，Flink的容错机制更高效，因为Flink的操作是对小批量数据而不是每条数据记录。但也不要让自己糊涂了，Flink仍然是原生流处理框架，它与Spark Streaming在概念上就完全不同。Flink也提供exactly once消息传输机制。

2. 状态管理

大部分大型流处理应用都涉及到状态。相对于无状态的操作(其只有一个输入数据，处理过程和输出结果)，有状态的应用会有一个输入数据和一个状态信息，然后处理过程，接着输出结果和修改状态信息。因此，我们不得不管理状态信息，并持久化。我们期望一旦因某种原因失败，状态能够修复。状态修复有可能会出现小问题，它并不总是保证exactly once，有时也会出现消费多次，但这并不是我们想要的。

2.1 Apache Storm

我们知道，Storm提供at-least once的消息传输保障。那我们又该如何使用Trident做到exactly once的语义。概念上貌似挺简单，你只需要提交每条数据记录，但这显然不是那么高效。所以你会想到小批量的数据记录一起提交会优化。Trident定义了几个抽象来达到exactly once的语义，见下图，其中也会有些局限。

2.2 Spark Streaming

Spark Streaming是微批处理系统，它把状态信息也看做是一种微批量数据流。在处理每个微批量数据时，Spark加载当前的状态信息，接着通过函数操作获得处理后的微批量数据结果并修改加载过的状态信息。

2.3 Samza

Samza实现状态管理是通过Kafka来处理的。Samza有真实的状态操作，所以其任务会持有一个状态信息，并把状态改变的日志推送到Kafka。如果需要状态重建，可以很容易的从Kafka的topic重建。为了达到更快的状态管理，Samza也支持把状态信息放入本地key-value存储中，所以状态信息不必一直在Kafka中管理，见下图。不幸的是，Samza只提供at-least once语义，exactly once的支持也在计划中。

2.4 Apache Flink

Flink提供状态操作，和Samza类似。Flink提供两种类型的状态：一种是用户自定义状态；另外一种是窗口状态。如图，第一个状态是自定义状态，它和其它的的状态不相互作用。这些状态可以分区或者使用嵌入式Key-Value存储状态(参阅文容错和状态)。当然Flink提供exactly-once语义。下图展示Flink长期运行的三个状态。

3. 单词计数例子中的状态管理

单词计数的详细代码见上篇文章，这里仅关注状态管理部分。

让我们先看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"));

 ...

 }

在第九行代码中，我们通过调用persistentAggregate创建一个状态。其中参数Count存储单词数，如果你想从状态中处理数据，你必须创建一个数据流。从代码中也可以看出实现起来不方便。

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))

首先我们需要创建一个RDD来初始化状态（第二行代码），然后进行transformations（第五行和六行代码）。接着在第八行到十四行代码，我们定义函数来处理单词数状态。函数计算并更新状态，最后返回结果。第十六行和十七行代码，我们得到一个状态信息流，其中包含单词数。

接着我们看下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)))
   }
 }

首先在第三行代码定义状态，进行Key-Value存储，在第五行到八行代码初始化状态。接着在计算中使用，上面的代码已经很直白。

最后，讲下Flink使用简洁的API实现状态管理：

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))
    }
}

我们仅仅需要在第六行代码中调用mapwithstate函数，它有一个函数参数（函数有两个变量，第一个是单词，第二个是状态。然后返回处理的结果和新的状态）。

4. 流处理框架性能

这里所讲的性能主要涉及到的是延迟性和吞吐量。

对于延迟性来说，微批处理一般在秒级别，大部分原生流处理在百毫秒以下，调优的情况下Storm可以很轻松的达到十毫秒。

同时也要记住，消息传输机制保障，容错性和状态恢复都会占用机器资源。例如，打开容错恢复可能会降低10％到15％的性能，Storm可能降低70%的吞吐量。总之，天下没有免费的午餐。对于有状态管理，Flink会降低25%的性能，Spark Streaming降低50%的性能。

也要记住，各大流处理框架的所有操作都是分布式的，通过网络发送数据是相当耗时的，所以要利用数据本地性，也尽量优化你的应用的序列化。

5. 项目成熟度

当你为应用选型时一定会考虑项目的成熟度。下面来快速浏览一下： Storm是第一个主流的流处理框架，后期已经成为长期的工业级的标准，并在像Twitter，Yahoo，Spotify等大公司使用。Spark Streaming是最近最流行的Scala代码实现的流处理框架。现在Spark Streaming被公司（Netflix, Cisco, DataStax, Intel, IBM等）日渐接受。Samza主要在LinkedIn公司使用。Flink是一个新兴的项目，很有前景。

你可能对项目的贡献者数量也感兴趣。Storm和Trident大概有180个代码贡献者；整个Spark有720多个；根据github显示，Samza有40个；Flink有超过130个代码贡献者。

6. 小结

在进行流处理框架推荐之前，先来整体看下总结表：

7. 流处理框架推荐

应用选型是大家都会遇到的问题，一般是根据应用具体的场景来选择特定的流处理框架。下面给出几个作者认为优先考虑的点：

• High level API：具有high level API的流处理框架会更简洁和高效；
• 状态管理：大部分流处理应用都涉及到状态管理，因此你得把状态管理作为评价指标之一；
• exactly once语义：exactly once会使得应用开发变得简单，但也要看具体需求，可能at least once或者at most once语义就满足你得要求；
• 自动恢复：确保流处理系统能够快速恢复，你可以使用Chaos Monkey或者类似的工具进行测试。快速的恢复是流处理重要的部分。

Storm：Storm非常适合任务量小但速度要求高的应用。如果你主要在意流处理框架的延迟性，Storm将可能是你的首先。但同时也要记住，Storm的容错恢复或者Trident的状态管理都会降低整体的性能水平。也有一个潜在的Storm更新项目-Twitter的Heron，Heron设计的初衷是为了替代Storm，并在每个单任务上做了优化但同时保留了API。

Spark Streaming：如果你得基础架构中已经涉及到Spark，那Spark Streaming无疑是值得你尝试的。因为你可以很好的利用Spark各种library。如果你需要使用Lambda架构，Spark Streaming也是一个不错的选择。但你要时刻记住微批处理的局限性，以及它的延迟性问题。

Samza：如果你想使用Samza，那Kafka应该是你基础架构中的基石，好在现在Kafka已经成为家喻户晓的组件。像前面提到的，Samza一般会搭配强大的本地存储一起，这对管理大数据量的状态非常有益。它可以轻松处理上万千兆字节的状态信息，但要记住Samza只支持at least once语义。

Flink：Flink流处理系统的概念非常不错，并且满足绝大多数流处理场景，也经常提供前沿的功能函数，比如，高级窗口函数或者时间处理功能，这些在其它流处理框架中是没有的。同时Flink也有API提供给通用的批处理场景。但你需要足够的勇气去上线一个新兴的项目，并且你也不能忘了看下Flink的roadmap。

8. Dataflow和开源

最后，我们来聊下Dataflow和它的开源。Dataflow是Google云平台的一部分，Google云平台包含很多组件：大数据存储，BigQuery，Cloud PubSub，数据分析工具和前面提到的Dataflow。

Dataflow是Google管理批处理和流处理的统一API。它是建立在MapReduce（批处理），FlumeJava（编程模型）和MillWheel（流处理）之上。Google最近决定开源Dataflow SDK，并完成Spark和Flink的runner。现在可以通过Dataflow的API来定义Google云平台作业、Flink作业或者Spark作业，后续会增加对其它引擎的支持。

Google为Dataflow提供Java、Python的API，社区已经完成Scalable的DSL支持。除此之外，Google及其合作者提交Apache Beam到Apache。

原文:http://www.infoq.com/cn/articles/comparison-of-main-stream-processing-framework-part02?utm_source=infoq&utm_campaign=user_page&utm_medium=link