Apache Beam: 下一代的大数据处理标准

Apache Beam（原名Google DataFlow）是Google在2016年2月份贡献给Apache基金会的Apache孵化项目，被认为是继MapReduce，GFS和BigQuery等以后，Google在大数据处理领域对开源社区的又一个很是大的贡献。Apache Beam的主要目标是统一批处理和流处理的编程范式，为无限，乱序，web-scale的数据集处理提供简单灵活，功能丰富以及表达能力十分强大的SDK。Apache Beam项目重点在于数据处理的编程范式和接口定义，并不涉及具体执行引擎的实现，Apache Beam但愿基于Beam开发的数据处理程序能够执行在任意的分布式计算引擎上。本文主要介绍Apache Beam的编程范式-Beam Model，以及经过Beam SDK如何方便灵活的编写分布式数据处理业务逻辑，但愿读者可以经过本文对Apache Beam有初步的了解，同时对于分布式数据处理系统如何处理乱序无限数据流的能力有初步的认识。

Apache Beam基本架构

随着分布式数据处理不断发展，新的分布式数据处理技术也不断被提出，业界涌现出了愈来愈多的分布式数据处理框架，从最先的Hadoop MapReduce，到Apache Spark，Apache Storm，以及更近的Apache Flink，Apache Apex等。新的分布式处理框架可能带来的更高的性能，更强大的功能，更低的延迟等，但用户切换到新的分布式处理框架的代价也很是大：须要学习一个新的数据处理框架，并重写全部的业务逻辑。解决这个问题的思路包括两个部分，首先，须要一个编程范式，可以统一，规范分布式数据处理的需求，例如，统一批处理和流处理的需求。其次，生成的分布式数据处理任务应该可以在各个分布式执行引擎上执行，用户能够自由切换分布式数据处理任务的执行引擎与执行环境。Apache Beam正是为了解决以上问题而提出的。

Apache Beam主要由Beam SDK和Beam Runner组成，Beam SDK定义了开发分布式数据处理任务业务逻辑的API接口，生成的的分布式数据处理任务Pipeline交给具体的Beam Runner执行引擎。Apache Beam目前支持的API接口是由Java语言实现的，Python版本的API正在开发之中。Apache Beam支持的底层执行引擎包括Apache Flink，Apache Spark以及Google Cloud Platform，此外Apache Storm，Apache Hadoop，Apache Gearpump等执行引擎的支持也在讨论或开发当中。其基本架构以下图所示：

图1 Apache Beam架构图

须要注意的是，虽然Apache Beam社区很是但愿全部的Beam执行引擎都可以支持Beam SDK定义的功能全集，可是在实际实现中可能并不必定。例如，基于MapReduce的Runner显然很难实现和流处理相关的功能特性。目前Google DataFlow Cloud是对Beam SDK功能集支持最全面的执行引擎，在开源执行引擎中，支持最全面的则是Apache Flink。

Beam Model

Beam Model指的是Beam的编程范式，即Beam SDK背后的设计思想。在介绍Beam Model以前，先简要介绍一下Beam Model要处理的问题域与一些基本概念。

1. 数据。分布式数据处理要处理的数据类型通常能够分为两类，有限的数据集和无限的数据流。有限的数据集，好比一个HDFS中的文件，一个HBase表等，特色是数据提早已经存在，通常也已经持久化，不会忽然消失。而无限的数据流，好比kafka中流过来的系统日志流，或是从twitter API拿到的twitter流等等，这类数据的特色是，数据动态流入，无穷无尽，没法所有持久化。通常来讲，批处理框架的设计目标是用来处理有限的数据集，流处理框架的设计目标是用来处理无限的数据流。有限的数据集能够看作是无限的数据流的一种特例，可是从数据处理逻辑的角度，这二者并没有不一样之处，例如，假设微博数据包含时间戳和转发量，用户但愿按照统计每小时的转发量总和，此业务逻辑应该能够同时在有限数据集和无限数据流上执行，并不该该由于数据源的不一样而对业务逻辑的实现产生任何影响。
2. 时间。Process Time是指数据进入分布式处理框架的时间，而Event-Time则是指数据产生的时间。这两个时间一般是不一样的，例如，对于一个处理微博数据的流计算任务，一条2016-06-01-12:00:00发表的微博通过网络传输等延迟可能在2016-06-01-12:01:30才进入到流处理系统中。批处理任务一般进行全量的数据计算，较少关注数据的时间属性，可是对于流处理任务来讲，因为数据流是无情无尽的，没法进行全量的计算，一般是对某个窗口中得数据进行计算，对于大部分的流处理任务来讲，按照时间进行窗口划分，多是最多见的需求。
3. 乱序。对于流处理框架处理的数据流来讲，其数据的到达顺序可能并不严格按照Event-Time的时间顺序。若是基于Process Time定义时间窗口，数据到达的顺序就是数据的顺序，所以不存在乱序问题。可是对于基于Event Time定义的时间窗口来讲，可能存在时间靠前的消息在时间靠后的消息后到达的状况，这在分布式的数据源中可能很是常见。对于这种状况，如何肯定迟到数据，以及对于迟到数据如何处理一般是很棘手的问题。

 

Beam Model处理的目标数据是无限的时间乱序数据流，不考虑时间顺序或是有限的数据集可看作是无限乱序数据流的一个特例。Beam Model从下面四个维度概括了用户在进行数据处理的时候须要考虑的问题：

1. What。如何对数据进行计算？例如，Sum，Join或是机器学习中训练学习模型等。在Beam SDK中由Pipeline中的操做符指定。
2. Where。数据在什么范围中计算？例如，基于Process-Time的时间窗口，基于Event-Time的时间窗口，滑动窗口等等。在BeamSDK中由Pipeline中的窗口指定。
3. When。什么时候将计算结果输出？例如，在1小时的Event-Time时间窗口中，每隔1分钟，将当前窗口计算结果输出。在Beam SDK中由Pipeline中的Watermark和触发器指定。
4. How。迟到数据如何处理？例如，将迟到数据计算增量结果输出，或是将迟到数据计算结果和窗口内数据计算结果合并成全量结果输出。在Beam SDK中由Accumulation指定。

Beam Model将”WWWH“四个维度抽象出来组成了Beam SDK，用户在基于Beam SDK构建数据处理业务逻辑时，在每一步只须要根据业务需求按照这四个维度调用具体的API便可生成分布式数据处理Pipeline，并提交到具体执行引擎上执行。“WWWH”四个维度的抽象仅仅关注业务逻辑自己，和分布式任务如何执行没有任何关系。

Beam SDK

不一样于Apache Flink或是Apache Spark，Beam SDK使用同一套API表示数据源，输出目标以及操做符等。下面介绍4个基于Beam SDK的数据处理任务，经过这四个数据处理任务，读者能够了解经过Beam Mode是如何统一灵活的描述批处理和流处理任务的，这4个任务用来处理手机游戏领域的统计需求，包括：

1. 用户分数。批处理任务，基于有限数据集统计用户分数。
2. 每小时团队分数。批处理任务，基于有限数据集统计每小时，每一个团队的分数。
3. 排行榜。流处理任务，2个统计项，每小时每一个团队的分数以及用户实时的历史总得分数。
4. 游戏状态。流处理任务，统计每小时每一个团队的分数，以及更复杂的每小时统计信息，好比每小时每一个用户在线时间等。

注：示例代码来自Beam的源码，具体地址参见：apache/incubator-beam。部分分析内容参考了Beam的官方文档，详情请参见引用连接。

下面基于Beam Model的“WWWH”四个维度，分析业务逻辑，并经过代码展现如何经过Beam SDK实现“WWWH”四个维度的业务逻辑。

用户分数

统计每一个用户的历史总得分数是一个很是简单的任务，在这里咱们简单的经过一个批处理任务实现，每次须要新的用户分数数据的时候，从新执行一次这个批处理任务便可。对于用户分数任务，“WWWH”四维度分析结果以下：

经过“WWWH”的分析，对于用户分数这个批处理任务，经过Beam Java SDK实现的代码以下所示：

gameEvents

  [... input ...]

    [... parse ...]

      .apply("ExtractUserScore", new ExtractAndSumScore("user")) 

  [... output ...];

ExtractAndSumScore实现了“What”中描述的逻辑，即按用户分组，而后累加分数，其相关代码以下：

gameInfo

 .apply(MapElements

     .via((GameActionInfo gInfo) -> KV.of(gInfo.getKey(field), gInfo.getScore()))

     .withOutputType(

         TypeDescriptors.kvs(TypeDescriptors.strings(), TypeDescriptors.integers())))

 .apply(Sum.<String>integersPerKey());

经过MapElements肯定Key与Value分别是用户与分数，而后Sum定义按key分组，并累加分数。Beam支持将多个对数据的操做合并成一个操做，这样不只能够支持更清晰的业务逻辑实现，同时也能够在多处重用合并后的操做逻辑。

每小时团队分数

按照小时统计每一个团队的分数，得到最高分数的团队可能得到奖励，这个分析任务增长了对窗口的要求，不过咱们依然能够经过一个批处理任务实现，对于这个任务的“WWWH”四个维度的分析以下：

相对于第一个用户分数任务，只是在Where部分回答了“数据在什么范围中计算？”的问题，同时在What部分“如何计算数据？”中，分组的条件由用户改成了团队，这在代码中也会相应的体现：

gameEvents

  [... input ...]

  [... parse ...]

  .apply("AddEventTimestamps", WithTimestamps.of((GameActionInfo i)

    -> new Instant(i.getTimestamp())))

  .apply("FixedWindowsTeam", Window.<GameActionInfo>into(

    FixedWindows.of(Duration.standardMinutes(windowDuration))))

  .apply("ExtractTeamScore", new ExtractAndSumScore("team"))

  [... output ...];

“AddEventTimestamps”定义了如何从原始数据中抽取EventTime数据，“FixedWindowsTeam”则定义了1小时固定窗口，而后重用了ExtractAndSumScore类，只是将分组的列从用户改为了团队。对于每小时团队分数任务，引入了关于“Where”部分窗口定义的新业务逻辑，可是从代码中能够看到，关于“Where”部分的实现和关于“What”部分的实现是彻底独立的，用户只须要新加两行关于“Where”的代码，很是简单和清晰。

排行榜

前面两个任务均是基于有限数据集的批处理任务，对于排行榜来讲，咱们一样须要统计用户分数以及每小时团队分数，可是从业务角度但愿获得的是实时数据。对于Apache Beam来讲，一个相同处理逻辑的批处理任务和流处理任务的惟一不一样就是任务的输入和输出，中间的业务逻辑Pipeline无需任何改变。对于当前示例的排行榜数据分析任务，咱们不只但愿他们知足和前两个示例相同的业务逻辑，同时也能够知足更定制化的业务需求，例如：

1. 流处理任务相对于批处理任务，一个很是重要的特性是，流处理任务能够更加实时的返回计算结果，例如计算每小时团队分数时，对于一小时的时间窗口，默认是在一小时的数据所有到达后，把最终的结算结果输出，可是流处理系统应该同时支持在一小时窗口只有部分数据到达时，就将部分计算结果输出，从而使得用户能够获得实时的分析结果。
2. 保证和批处理任务一致的计算结果正确性。因为乱序数据的存在，对于某一个计算窗口，如何肯定全部数据是否到达（Watermark）？迟到数据如何处理？处理结果如何输出，总量，增量，并列？流处理系统应该提供机制保证用户能够在知足低延迟性能的同时达到最终的计算结果正确性。

上述两个问题正是经过回答“When”和“How”两个问题来定义用户的数据分析需求。“When”取决于用户但愿多常获得计算结果，在回答“When”的时候，基本上能够分为四个阶段：

1. Early。在窗口结束前，肯定什么时候输出中间状态数据。
2. On-Time。在窗口结束时，输出窗口数据计算结果。因为乱序数据的存在，如何判断窗口结束多是用户根据额外的知识预估的，且容许在用户设定的窗口结束后出现迟到的属于该窗口的数据。
3. Late。在窗口结束后，有迟到的数据到达，在这个阶段，什么时候输出计算结果。
4. Final。可以容忍迟到的最大限度，例如1小时。到达最后的等待时间后，输出最终的计算结果，同时再也不接受以后的迟到数据，清理该窗口的状态数据。

对于每小时团队得分的流处理任务，本示例但愿的业务逻辑为，基于Event Time的1小时时间窗口，按团队计算分数，在一小时窗口内，每5分钟输出一次当前的团队分数，对于迟到的数据，每10分钟输出一次当前的团队分数，在窗口结束2小时后迟到的数据通常不可能会出现，假如出现的话，直接抛弃。“WWWH”表达以下：

在基于Beam SDK的实现中，用户基于“WWWH” Beam Model表示的业务逻辑能够分别独立直接的实现出来：

gameEvents
 [... input ...]
  .apply("LeaderboardTeamFixedWindows", Window
    .<GameActionInfo>into(FixedWindows.of(
      Duration.standardMinutes(Durations.minutes(60))))
    .triggering(AfterWatermark.pastEndOfWindow()
      .withEarlyFirings(AfterProcessingTime.pastFirstElementInPane()
        .plusDelayOf(Durations.minutes(5)))
      .withLateFirings(AfterProcessingTime.pastFirstElementInPane()
        .plusDelayOf(Durations.minutes(10))))
    .withAllowedLateness(Duration.standardMinutes(120)
    .accumulatingFiredPanes())
  .apply("ExtractTeamScore", new ExtractAndSumScore("team"))
  [... output ...]

LeaderboardTeamFixedWindows对应“Where”定义窗口，Trigger对应“Where”定义结果输出条件，Accumulation对应“How”定义输出结果内容，ExtractTeamScore对应“What”定义计算逻辑。

总结

Apache Beam的Beam Model对无限乱序数据流的数据处理进行了很是优雅的抽象，“WWWH”四个维度对数据处理的描述，很是清晰与合理，Beam Model在统一了对无限数据流和有限数据集的处理模式的同时，也明确了对无限数据流的数据处理方式的编程范式，扩大了流处理系统可应用的业务范围，例如，Event-Time/Session窗口的支持，乱序数据的处理支持等。Apache Flink，Apache Spark Streaming等项目的API设计均愈来愈多的借鉴或参考了Apache Beam Model，且做为Beam Runner的实现，与Beam SDK的兼容度也愈来愈高。本文主要介绍了Beam Model，以及如何基于Beam Model设计现实中的数据处理任务，但愿可以让读者对Apache Beam项目可以有一个初步的了解。因为Apache Beam已经进入Apache Incubator孵化，因此读者也能够经过官网或是邮件组了解更多Apache Beam的进展和状态。

大数据学习交流资本群：784557197