BT

如何利用碎片时间提升技术认知与能力? 点击获取答案

主流流处理框架比较(系列文章)

| 作者 Petr Zapletal 关注 0 他的粉丝 ,译者 侠天 关注 5 他的粉丝 发布于 2016年7月8日. 估计阅读时间: 18 分钟 | QCon上海2018 关注大数据平台技术选型、搭建、系统迁移和优化的经验。

分布式流处理是对无边界数据集进行连续不断的处理、聚合和分析。它跟MapReduce一样是一种通用计算,但我们期望延迟在毫秒或者秒级别。这类系统一般采用有向无环图(DAG)。

DAG是任务链的图形化表示,我们用它来描述流处理作业的拓扑。如下图,数据从sources流经处理任务链到sinks。单机可以运行DAG,但本篇文章主要聚焦在多台机器上运行DAG的情况。

关注点

当选择不同的流处理系统时,有以下几点需要注意的:

  • 运行时和编程模型:平台框架提供的编程模型决定了许多特色功能,编程模型要足够处理各种应用场景。这是一个相当重要的点,后续会继续。
  • 函数式原语:流处理平台应该能提供丰富的功能函数,比如,map或者filter这类易扩展、处理单条信息的函数;处理多条信息的函数aggregation;跨数据流、不易扩展的操作join。
  • 状态管理:大部分应用都需要保持状态处理的逻辑。流处理平台应该提供存储、访问和更新状态信息。
  • 消息传输保障:消息传输保障一般有三种:at most once,at least once和exactly once。At most once的消息传输机制是每条消息传输零次或者一次,即消息可能会丢失;A t least once意味着每条消息会进行多次传输尝试,至少一次成功,即消息传输可能重复但不会丢失;Exactly once的消息传输机制是每条消息有且只有一次,即消息传输既不会丢失也不会重复。
  • 容错:流处理框架中的失败会发生在各个层次,比如,网络部分,磁盘崩溃或者节点宕机等。流处理框架应该具备从所有这种失败中恢复,并从上一个成功的状态(无脏数据)重新消费。
  • 性能:延迟时间(Latency),吞吐量(Throughput)和扩展性(Scalability)是流处理应用中极其重要的指标。
    平台的成熟度和接受度:成熟的流处理框架可以提供潜在的支持,可用的库,甚至开发问答帮助。选择正确的平台会在这方面提供很大的帮助。

运行时和编程模型

运行时和编程模型是一个系统最重要的特质,因为它们定义了表达方式、可能的操作和将来的局限性。因此,运行时和编程模型决定了系统的能力和适用场景。

实现流处理系统有两种完全不同的方式:一种是称作原生流处理,意味着所有输入的记录一旦到达即会一个接着一个进行处理。

第二种称为微批处理。把输入的数据按照某种预先定义的时间间隔(典型的是几秒钟)分成短小的批量数据,流经流处理系统。

两种方法都有其先天的优势和不足。首先以原生流处理开始,原生流处理的优势在于它的表达方式。数据一旦到达立即处理,这些系统的延迟性远比其它微批处理要好。除了延迟性外,原生流处理的状态操作也容易实现,后续将详细讲解。一般原生流处理系统为了达到低延迟和容错性会花费比较大的成本,因为它需要考虑每条记录。原生流处理的负载均衡也是个问题。比如,我们处理的数据按key分区,如果分区的某个key是资源密集型,那这个分区很容易成为作业的瓶颈。

接下来看下微批处理。将流式计算分解成一系列短小的批处理作业,也不可避免的减弱系统的表达力。像状态管理或者join等操作的实现会变的困难,因为微批处理系统必须操作整个批量数据。并且,batch interval会连接两个不易连接的事情:基础属性和业务逻辑。相反地,微批处理系统的容错性和负载均衡实现起来非常简单,因为微批处理系统仅发送每批数据到一个worker节点上,如果一些数据出错那就使用其它副本。微批处理系统很容易建立在原生流处理系统之上。

编程模型一般分为组合式和声明式。组合式编程提供基本的构建模块,它们必须紧密结合来创建拓扑。新的组件经常以接口的方式完成。相对应地,声明式API操作是定义的高阶函数。它允许我们用抽象类型和方法来写函数代码,并且系统创建拓扑和优化拓扑。声明式API经常也提供更多高级的操作(比如,窗口函数或者状态管理)。后面很快会给出样例代码。

主流流处理系统

有一系列各种实现的流处理框架,不能一一列举,这里仅选出主流的流处理解决方案,并且支持Scala API。因此,我们将详细介绍Apache Storm,Trident,Spark Streaming,Samza和Apache Flink。前面选择讲述的虽然都是流处理系统,但它们实现的方法包含了各种不同的挑战。这里暂时不讲商业的系统,比如Google MillWheel或者Amazon Kinesis,也不会涉及很少使用的Intel GearPump或者Apache Apex。

Apache Storm最开始是由Nathan Marz和他的团队于2010年在数据分析公司BackType开发的,后来BackType公司被Twitter收购,接着Twitter开源Storm并在2014年成为Apache顶级项目。毋庸置疑,Storm成为大规模流数据处理的先锋,并逐渐成为工业标准。Storm是原生的流处理系统,提供low-level的API。Storm使用Thrift来定义topology和支持多语言协议,使得我们可以使用大部分编程语言开发,Scala自然包括在内。

Trident是对Storm的一个更高层次的抽象,Trident最大的特点以batch的形式进行流处理。Trident简化topology构建过程,增加了窗口操作、聚合操作或者状态管理等高级操作,这些在Storm中并不支持。相对应于Storm的At most once流传输机制,Trident提供了Exactly once传输机制。Trident支持Java,Clojure和Scala。

当前Spark是非常受欢迎的批处理框架,包含Spark SQL,MLlib和Spark Streaming。Spark的运行时是建立在批处理之上,因此后续加入的Spark Streaming也依赖于批处理,实现了微批处理。接收器把输入数据流分成短小批处理,并以类似Spark作业的方式处理微批处理。Spark Streaming提供高级声明式API(支持Scala,Java和Python)。

Samza最开始是专为LinkedIn公司开发的流处理解决方案,并和LinkedIn的Kafka一起贡献给社区,现已成为基础设施的关键部分。Samza的构建严重依赖于基于log的Kafka,两者紧密耦合。Samza提供组合式API,当然也支持Scala。

最后来介绍Apache Flink。Flink是个相当早的项目,开始于2008年,但只在最近才得到注意。Flink是原生的流处理系统,提供high level的API。Flink也提供API来像Spark一样进行批处理,但两者处理的基础是完全不同的。Flink把批处理当作流处理中的一种特殊情况。在Flink中,所有的数据都看作流,是一种很好的抽象,因为这更接近于现实世界。

快速的介绍流处理系统之后,让我们以下面的表格来更好清晰的展示它们之间的不同:

Word Count

Wordcount之于流处理框架学习,就好比hello world之于编程语言学习。它能很好的展示各流处理框架的不同之处,让我们从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<String, Integer> counts = new HashMap<String, Integer>();

 public void execute(Tuple tuple, BasicOutputCollector collector) {
   String word = tuple.getString(0);
   Integer count = counts.containsKey(word) ? counts.get(word) + 1 : 1;
   counts.put(word, count);
   collector.emit(new Values(word, count));
 }

首先,定义topology。第二行代码定义一个spout,作为数据源。然后是一个处理组件bolt,分割文本为单词。接着,定义另一个bolt来计算单词数(第四行代码)。也可以看到魔数5,8和12,这些是并行度,定义集群每个组件执行的独立线程数。第八行到十五行是实际的WordCount bolt实现。因为Storm不支持内建的状态管理,所有这里定义了一个局部状态。

按之前描述,Trident是对Storm的一个更高层次的抽象,Trident最大的特点以batch的形式进行流处理。除了其它优势,Trident提供了状态管理,这对wordcount实现非常有用。

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

 ...

 }

如你所见,上面代码使用higher level操作,比如each(第七行代码)和groupby(第八行代码)。并且使用Trident管理状态来存储单词数(第九行代码)。

下面是时候祭出提供声明式API的Apache 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 Streaming的作业都要有StreamingContext,它是流式函数的入口。StreamingContext加载第一行代码定义的配置conf,但更重要地,第二行代码定义batch interval(这里设置为1秒)。第六行到八行代码是整个单词数计算。这些是标准的函数式代码,Spark定义topology并且分布式执行。第十二行代码是每个Spark Streaming作业最后的部分:启动计算。记住,Spark Streaming作业一旦启动即不可修改。
接下来看下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))

 }

Samza的属性配置文件定义topology,为了简明这里并没把配置文件放上来。定义任务的输入和输出,并通过Kafka topic通信。在单词数计算整个topology是WordCountTask。在Samza中,实现特殊接口定义组件StreamTask,在第三行代码重写方法process。它的参数列表包含所有连接其它系统的需要。第八行到十行简单的Scala代码是计算本身。
Flink的API跟Spark Streaming是惊人的相似,但注意到代码里并未设置batch interval。

val env = ExecutionEnvironment.getExecutionEnvironment

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

 counts.print()

 env.execute("wordcount")

上面的代码是相当的直白,仅仅只是几个函数式调用,Flink支持分布式计算。

结论

上面给出了基本的理论和主流流处理框架介绍,下篇文章将会更深入的探讨其它关注点。希望你能对前面的文章感兴趣,如果有任何问题,请联系我讨论这些主题。

译者介绍

侠天,专注于大数据、机器学习和数学相关的内容,并有个人公众号:bigdata_ny分享相关技术文章。

英文原文:Comparison of Apache Stream Processing Frameworks: Part 1

评价本文

专业度
风格

您好,朋友!

您需要 注册一个InfoQ账号 或者 才能进行评论。在您完成注册后还需要进行一些设置。

获得来自InfoQ的更多体验。

告诉我们您的想法

允许的HTML标签: a,b,br,blockquote,i,li,pre,u,ul,p

当有人回复此评论时请E-mail通知我

翻译建议 by Z DeepLearning

翻译不到位:“声明式API操作是定义的高阶函数。它允许我们用抽象类型和方法来写函数代码,并且系统创建拓扑和优化拓扑”这句原文是:operators in declarative API are defined as higher order functions. It allows us to write functional code with abstract types and all its fancy stuff and the system creates and optimizes topology itself.
应翻译为:dsl API都以高阶函数形式定义,就像scala集合库方法....[由]系统[来]创建和优化拓扑。
潜台词:你只要写功能,怎么运行你不用管了
SparkStreaming只具备这一种方式,和其他Spark模块一致,因为Spark后台都是以DAG形式,这个DAG由Spark自己管理,从SparkContext衍生出来看上去只是普通的scala代码调用,后台Spark是做了处理的:转化为分布式DAG

允许的HTML标签: a,b,br,blockquote,i,li,pre,u,ul,p

当有人回复此评论时请E-mail通知我

允许的HTML标签: a,b,br,blockquote,i,li,pre,u,ul,p

当有人回复此评论时请E-mail通知我

1 讨论

登陆InfoQ,与你最关心的话题互动。


找回密码....

Follow

关注你最喜爱的话题和作者

快速浏览网站内你所感兴趣话题的精选内容。

Like

内容自由定制

选择想要阅读的主题和喜爱的作者定制自己的新闻源。

Notifications

获取更新

设置通知机制以获取内容更新对您而言是否重要

BT