Dataflow编程模型和spark streaming结合
主要介绍一下Dataflow编程模型的基本思想，后面再简单比较一下Spark streaming的编程模型
== 是什么 ==
为用户提供以流式或批量模式处理海量数据的能力，该服务的编程接口模型（或者说计算框架）也就是下面要讨论的dataflow model
流式计算框架处理框架很多，也有大量的模型／框架号称能较好的处理流式和批量计算场景，比如Lambda模型，比如Spark等等，那么dataflow模型有什么特别的呢？
这就要要从流式（streaming） 和批量（ batch） 这两个词的语意说起，简单的说，谷歌的同学认为目前的各种计算框架模型以及用户在一定程度上对这两个词的语意的使用姿势不够恰当，或者说用这两个词来区分应用场景，进而给计算框架分类并不合适。而这种分类给框架的应用甚至设计带来了一定的认识的误解和偏差。
同学认为，当大家谈论到流式计算，或者流式数据时，内心想表达的场景，实际上更准确的说法应该是unbounded data(processing)，也就是无边界的连续的数据（的处理）；对应的批量（计算），更准确的说法是bounded data(processing)，亦即有明确边界的数据的处理。而stream和batch实际只是这两种数据集历史上传统使用的处理方式而已，这两者并不完全等价。而随着技术的发展，继续用这种方式来分类和看待问题就显得不够高大上了。
一篇chanllenge前段时间热门的Lambda模型 https://www.oreilly.com/ideas/questioning-the-lambda-architecture 的文章中所表达的观点，一定程度上也是上面这种思想的一个体现。
而Dataflow模型则是谷歌的同学在处理无边界数据的实践中，总结的一套SDK级别的解决方案，其目标是做到在非有序的，无边界的海量数据上，基于事件时间进行运算，并能根据数据自身的属性进行window操作，同时数据处理过程的正确性，延迟，代价可根据需求进行灵活的调整配置。其底层计算引擎依托于 Millwheel 实时计算框架和FlumeJava批处理框架，在开源了相关SDK以后，发起了beam项目（http://beam.incubator.apache.org/），其底层计算引擎也可以替换适配成Spark／Flink等开源计算框架（进行中）
== 核心思想 ==
基本上Lambda模型被挑战的点是，用一个流式＋批量的拼凑方案去解决海量无限数据的实时统计问题，看起来很美，但是出发点立意有些Low（亦即，认定了这种问题只能通过两套截然不同的框架模型去协同处理），而维护两套计算框架模型和处理逻辑的代价始终是这个模型无法克服的痛点。虽然有各种在上层封装抽象统一的SDK编程接口方案的存在，企图通过一套代码，翻译执行的方式，降低在两套计算框架模型上开发和维护代码的代价，但实际效果往往并不如意，翻译执行层的存在，并不能抹平两种计算框架模型的差异，到头来真正能复用的代码逻辑并不多，简单的说就是Lambada框架本身并不解决用户真正的痛点，只是一种无奈之举。
而Dataflow计算模型，则是希望从编程模型的源头上就统一解决传统的流式和批量两种计算语意希望处理的问题。
和spark通过micro batch来处理streaming场景（如前，更准确的说法是无边界数据集）的出发点不同，Dataflow认为batch的处理模式只是streamming处理模式的一个子集。在无边界数据集的处理过程中，要及时产出数据结果，必然需要对需要处理的数据划定一个窗口区间，从而对数据及时的进行分段处理和产出，各种处理模式（stream，micro batch，session， batch）只是窗口的大小不同，窗口的划分方式不同，比如，Batch的处理模式就只是一个窗口区间涵盖了整个有边界的数据集这样的一种特例而已。一个设计良好的能处理无边界数据集的系统，完全能在准确性和正确性上做到和“Batch”系统一样甚至应该更好。而不是传统的认为batch框架的正确性更好，streaming框架顾及了实时性，正确性天然就做不好，必须和batch框架配合走Lambada模型来补足。
那么无边界数据集的处理过程中，大家认为天然做不好的点，或者说最难处理的点在哪里，Dataflow是怎么解决的呢。
这里又要先说一下在Dataflow模型里强调的两个时间概念：Event time和 process time，Event time 事件时间就是数据真正发生的时间，比如用户浏览了一个页面，或者下了一个订单等等，这时候通畅就会有一些数据会被生产出来，比如前者可能会产生一条用户的浏览日志，而process time则是这条日志数据真正到达计算框架中被处理的时间点。现实情况下，由于各种原因，数据采集，传输到达处理系统的时间可能会有长短不同的延迟，在分布式应用场景环境下，不仅是延迟，数据乱序到达往往也是常态。这些问题，在有边界数据的处理场景过程中往往并不存在，或者无关紧要。
基于这种无边界数据集的特性，在Dataflow模型中，数据的处理过程被概括为以下4个方面的问题的解决：
– What results are being computed. ： 计算逻辑是什么
– Where in event time they are being computed. ： 计算什么时候（事件时间）的数据
– When in processing time they are materialized. ： 在什么时候（处理时间）进行计算
– How earlier results relateto later refinements. ： 后续数据的处理结果如何影响之前的处理结果
清晰的定义这些问题，并针对性的在模型框架层面加以解决，正是Dataflow区别于其它流式计算模型的核心关键所在。通常的流式计算框架往往模糊或者无法有效的区别对待数据的事件时间和处理时间，对于第4个问题，也可能缺乏直接的支持。这些问题通常需要开发人员自行在代码业务逻辑上想办法解决，因而也就加大了这类数据处理业务的开发难度，甚至成为一个不可能完成的任务。
而更重要的是，针对同一或类似数据集，各种数据处理需求，其核心计算逻辑往往可能是一致的，比如计算活跃用户数，核心计算逻辑就是一个去重逻辑。 但是根据应用目标场景，统计口径可能各有不同，比如可能要求计算过去一个小时的活跃用户，也可能是计算全天的累计的活跃用户，可能基于实际时间计算也可能基于数据采集时间计算，可能要求更新历史数据（有数据晚到），也可能处于效率，性能考虑，直接放弃晚到的数据。 Dataflow计算模型的目标是把上述4方面的问题，用明确的语意清晰的拆分出来，更好的模块化，快速适应各种业务逻辑开发需求。
例如在 https://cloud.google.com/dataflow/blog/dataflow-beam-and-spark-comparison 一文中，就用实际的例子比较了dataflow和spark在处理这类数据业务逻辑时，所需要进行的开发工作，总体的意思就是用dataflow模型开发，代码更简洁更容易理解，开发效率更高，维护成本更低。不过，需要注意的是，spark2.0的structure streaming API也引入了和Dataflow类似的模型思想，这篇文章里的很多比较已经不成立。
== 实现 ==
那么Dataflow是如何解决上面4方面的问题的呢，基本上，是通过构建以下三个核心功能模型来做到的：
- 一个支持基于事件时间的窗口（window）模型，并提供简易的API接口：支持固定窗口／滑动窗口／Session（以Key为维度，基于事件时间连续性进行划分）等窗口模式
- 一个和数据自身特性绑定的计算结果输出触发模型，并提供灵活可描述的API接口
- 一个增量更新模型，可以将数据增量更新的能力融合进上述窗口和结果触发模型中。
=== 窗口模型 ===
为了在计算框架级别实现基于事件时间的窗口模型，Dataflow系统中，将常见的流式计算框架中的［key，value］两元组tuple形式的信息数据，变换成了［key,value, event time, window ］这样的四元组模型，event time的引入原因显而易见，必须要有相关载体承载这个信息（否则只能基于process time／batch time 划分窗口），而window窗口标识信息的引入，个人认为，很重要的一个原因是要支持Session类型的窗口模型，而同时，要将流式和增量更新的支持融合进窗口的概念中，也势必需要在数据中引入这样一个显示的窗口信息（否则，通常的做法就只能是用micro batch分组数据的方式，隐式的标识数据的窗口属性）
在消息的四元组数据结构基础上，Dataflow通过提供对消息进行窗口赋值，窗口合并，按key分组，按窗口分组等原子功能操作，来实现各种窗口模型。
=== 触发模型 ===
多数的基于Process time的固定或滑动窗口模型，并没有显示的窗口计算结果触发这样一个概念的定义，因为不太需要，窗口的边界时间点，也就是触发结果输出的时间点。而对于Dataflow来说，因为事件时间和处理时间的延迟，以及框架需要正确处理无序数据的需求，使得判断窗口的边界，触发计算和结果的输出变得困难起来。在这一点上，Dataflow部分借用了底层Millwheel提供的Low watermark低水位这样一个概念来解决窗口边界的判断问题，当低水位对应的时间点超过设定的时间窗口边界时间点时，完成窗口的计算和结果输出。但是，低水位的概念理论上虽然是OK的，在实际场景中，通常是一个概率模型，并不能完全保证准确的判断事件时间的延迟情况，而且有很多场合对窗口边界的判断，用户自己有自己的需求。
因此，Dataflow提供了可自定义的窗口触发模型，可以使用低水位做触发，也可以使用比如：定时触发，计数触发，计量触发，模式匹配触发或其它外部触发源，甚至各种触发条件的逻辑运算组合等不同等机制来应对可能的需求。
=== 增量更新 ===
当窗口被触发以后，对于后续晚到的数据，对已经触发过的窗口，如何处理，Dataflow在框架层面也提供了直接的支持，基本上包括三种策略：
- 丢弃：一旦特定窗口触发过，对应窗口的数据就丢弃，晚到的数据也丢弃。
- 累积：触发过的窗口对应的数据保留（保留时间策略也可调整），晚到的数据更新对应窗口的输出结果
- 累计并更正：和累积模式类似，区别在于会先对上一次窗口触发的结果发送一个反相修正的信息，再输出新的结果，便于有需要的下游更正之前收到的信息。
== 相关研究，项目等 ==
=== spark 2.0 ===
Spark 2.0版本，新增的structured streaming API，针对原先的streaming编程接口DStream的问题进行了改进，Dstream的问题包括：
- 框架自身只能针对Batch time进行处理，很难处理event time，很难处理延迟，乱序的数据
- 流式和批量处理的API还是不完全一致，两种使用场景中，程序代码还是需要一定的转换
- 端到端的数据容错保障逻辑需要用户自己小心构建，增量更新和持久化存储等一致性问题处理难度较大
通过Structured Streaming API，Spark一方面支持了和Dataflow类似的概念，如Event time based的窗口策略，自定义的触发逻辑，对输出（sink）模块的更新模式（追加，全量覆盖，更新）的builtin支持，更加统一的处理无边界数据和有边界数据等。
总体看来，Spark 2.0的structured streaming 模型和Dataflow有异曲同工之处，设计的目标看起来很远大，甚至给出了一份功能比较表格来证明其优越性
不过在2.0的版本所支持的类Dataflow模型的功能还相对简单，比如session window，water flow等概念都还需要在2.1或者后续的版本中保证，也还不支持输出的更新模式，追加模式更新只能支持无聚合操作的场景，还有各种功能还停留在设想阶段，对于join等操作还有各种各样的限制等等，这些部分和dataflow业已实现的功能还有较大的差距。
对于exactly once发送的保障，spark2.0要求外部数据源具备offset定位的能力，再加上snapshot等机制来实现，而dataflow是通过对消息在框架内部进行持久化来实现replay，不依赖外部数据源的能力。
另外，个人理解像 prefix integrity， Transactional sink等概念，实际上是对上下游读写接口的一个封装，帮用户实现了一些业务逻辑（比如prefix integrity 的实现依托于于per key有序性的保证，这是由外部source源提供的保障，比如 file／kafka等；而Transactional sinks等则是比如对jdbc接口逻辑的封装），整体上偏外围功能一点，用这些特性来和其它框架比较不一定客观，因为设计理念不太不一样。Dataflow的模型设计中，用户能更加细化的定义每个环节的步骤和设置，所以不会把一些逻辑替用户实现，更多的是以模块化的方式，留给用户去自己选择，而Structured steaming则把很多事情包办了，定制的余地较小，灵活性应该会差一些，不过这也给程序的自动优化带来了一些便利。当然，这是我个人初步粗浅的理解，不见得准确。
=== beam ===
Beam http://beam.incubator.apache.org/ 是一个由谷歌发起的apache 项目，目前还处于incubator状态，基本来说就是实现dataflow编程模型的SDK项目，目标是提供一个high level的统一API编程接口，后端的执行引擎计划对接spark／flink／cloud dataflow。目前的编程语言支持Java，计划加入Python。这个项目的前景如何，不太好说，单就适配各个后端的角度来说，就spark后端来说，在spark 1.x时代，这种high level的编程模型抽象是对spark编程模型的一种add on，有一定的附加价值，但是按照spark 2.0 structured streaming的发展路线来说，这一层抽象就稍微显得有些多余了。而基于Java的语法，在表达的简洁性上，相比Scala也会带来一些额外的代价。
== 参考资料 ==
- dataflow论文 ： The Dataflow Model- A Practical Approach to Balancing Correctness, Latency, and Cost in Massive-Scale, Unbounded, Out-of-Order Data Processing
- https://cloud.google.com/dataflow/blog/dataflow-beam-and-spark-comparison
- https://spark-summit.org/2016/events/structuring-spark-dataframes-datasets-and-streaming/
- https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html
- https://databricks.com/blog/2016/07/28/continuous-applications-evolving-streaming-in-apache-spark-2-0.html
- https://www.oreilly.com/ideas/the-world-beyond-batch-streaming-101
- https://www.oreilly.com/ideas/the-world-beyond-batch-streaming-102
- https://cloud.google.com/dataflow
- http://beam.incubator.apache.org/