问题导读：
1.怎样优化Logical Plan？
2.怎样优化Stream Graph？
3.TimeWindow，EventTime，ProcessTime 和 Watermark四者之间的关系是什么？
序言
时效性提升数据的价值，所以Flink这样的流式(Streaming)计算系统应用得越来越广泛。
广大的普通用户决定一个产品的界面和接口。
ETL开发者需要简单而有效的开发工具，从而把更多时间花在理业务和对口径上。
因此流式计算系统都趋同以SQL作为唯一开发语言，让用户以Table形式操作Stream。
程序开发三部曲：First make it work, then make it right, and, finally, make it fast.
流计算开发者面对的现状及趋势：
第一步，让程序运行起来。
开发者能用SQL方便地表达问题。
开发者能通过任务管理系统一体化地管理任务，如：开发，上线，调优，监控和排查任务。
第二步，让程序运行正确。
简单数据清洗之外的流计算开发需求通常会涉及到Streaming SQL的两个核心扩展：Window 和 Emit。
开发者深入理解Window和 Emit的语义是正确实现这些业务需求的关键，
否则无法在数据时效性和数据准确性上做适合各个业务场景的决策和折中。
第三步，让程序运行越来越快。
苹果每年都会发布新手机：使用了**芯片，性能提升了多少，耗电降低了多少，增加**功能...。
当前，流计算系统每年也会有很大的性能提升和功能扩展，但想要深入调优及排错，
还是要学习分布式系统的各个组件及原理，各种算子实现方法，性能优化技术等知识。
以后，随着系统的进一步成熟和完善，开发者在性能优化上的负担会越来越低
无需了解底层技术实现细节和手动配置各种参数，就能享受性能和稳定性的逐步提升。
分布式系统的一致性和可用性是一对矛盾。
流计算系统的数据准确性和数据时效性也是一对矛盾。
应用开发者都需要认识到这些矛盾，并且知道自己在什么场景下该作何种取舍。
本文希望通过剖析Flink Streaming SQL的三个具体例子：Union，Group By 和 Join ，
来依次阐述流式计算模型的核心概念: What, Where, When, How 。
以便开发者加深对Streaming SQL的Window 和 Emit语义的理解，
从而能在数据准确性和数据时效性上做适合业务场景的折中和取舍。
也顺带介绍Streaming SQL的底层实现，以便于SQL任务的开发和调优。
UNION
通过这个例子来阐述Streaming SQL的底层实现和优化手段：Logical Plan Optimization 和 Operator Chaining。
例子
改编自Flink StreamSQLExample 。只在最外层加了一个Filter，以便触发Filter下推及合并。
Source
SQL
Sink
运行结果
转换Table为Stream
Flink 会把基于Table的Streaming SQL转为基于Stream的底层算子，并同时完成Logical Plan及Operator Chaining等优化
转为逻辑计划(Logical Plan)
上述UNION ALL SQL依据Relational Algebra转换为下面的逻辑计划:
SQL字段与逻辑计划有如下的对应关系：
优化Logical Plan理论基础幂等
数学: 19 * 10 * 1 * 1 = 19 * 10 = 190
SQL: SELECT * FROM (SELECT user, product FROM OrderA) = SELECT user, product FROM OrderA
数学：10 * 19 = 19 * 10 = 190
SQL: tableA UNION ALL tableB = tableB UNION ALL tableA
数学:
（1900 * 0.5）* 0.2 = 1900 * (0.5 * 0.2) = 190
1900 * (1.0 + 0.01) = 1900 * 1.0 + 1900 * 0.01 = 1919
SQL:
SELECT * FROM (SELECT user, amount FROM OrderA) WHERE amount > 2
SELECT * FROM (SELECT user, amount FROM OrderA WHERE amount > 2)
优化过程
Flink的逻辑计划优化规则清单请见: FlinkRuleSets
此Union All例子根据幂等，交换律和结合律来完成以下三步优化：
消除冗余的Project
利用幂等特性，消除冗余的Project
下推Filter
利用交换率和结合律特性，下推Filter。
合并Filter
利用结合律，合并Filter。
转为物理计划（Physical Plan）
转换后的Flink的物理执行计划如下：
有Physical Plan优化这一步骤，但对以上例子没有效果，所以忽略。
这样，加上Source和Sink，产生了如下的Stream Graph：
优化Stream Graph
通过Task Chaining来减少上下游算子的数据传输消耗，从而提高性能。
Chaining判断条件
Chaining结果
按深度优先的顺序遍历Stream Graph，最终产生5个Task任务。
GROUP BY
将以滚动窗口的GROUP BY来阐述Streaming SQL里的Window和Emit语义，
及其背后的Streaming的Where(Window)和When(Watermark和Trigger)的概念及关系。
例子 Source
Water Mark
简单地把最新的EventTime减去Offset。
SQL
按用户加滚动窗口进行Group By。
Sink
转换Table为Stream
因为Union All例子比较详细地阐述了转换规则，此处只讨论特殊之处。
转为逻辑计划(Logical Plan)
优化Logical Plan
GROUP BY优化：把{“User + Window” -> SUM} 转为 {User -> {Window -> SUM}}。
新的数据结构确保同一User下所有Window都会被分配到同一个Operator，以便实现SessionWindow的Merge功能。
转为物理计划（Physical Plan）
优化Stream Graph
经过Task Chaining优化后，最终生成3个Task。
Streaming各基本概念之间的联系
此处希望以图表的形式阐述各个概念之间的关系。
Window和EventTime
Flink支持三种Window类型: Tumbling Windows , Sliding Windows 和 Session Windows
每个事件的EventTime决定事件会落到哪些TimeWindow。
但只有Window的第一个数据来到时，Window才会被真正创建。
Window和WaterMark
可以设置TimeWindow的AllowedLateness，从而使Window可以处理延时数据。
只有当WaterMark超过TimeWindow.end + AllowedLateness时，Window才会被销毁。
TimeWindow，EventTime，ProcessTime 和 Watermark
我们以WaterMark的推进图来阐述这四者之间的关系。
Window为TumbleWindow，窗口大小为1小时，允许的数据延迟为1小时。
WaterMark和EventTime：
新数据的最新Eventime推进WaterMark。
TimeWindow的生命周期:
以下三条数据的EventTime决定TimeWindow的状态转换。
数据1的Eventtime属于Window[10:00, 11,00)，因为Window不存在，所以创建此Window。
数据2的Eventime推进WaterMark超过11:00(Window.end)，所以触发Pass End。
数据3的Eventime推进WaterMark超过12:00(Window.end + allowedLateness), 所以关闭此Window。
TimeWindow的结果输出：
用户可以通过Trigger来控制窗口结果的输出，按窗口的状态类型有以下三种Trigger。
Flink的Streaming SQL目前只支持PassEnd Trigger，且默认AllowedLateness = 0。
如果触发频率是Repeated，比如：每分钟, 往下游输出一次。那么这个时间只能是ProcessTime。
因为WarkMark在不同场景下会有不同推进速度，比如处理一小时的数据，
可能只需十分钟(重跑)，一个小时(正常运行）或 大于1小时（积压）。
运行结果
允许数据乱序是分布式系统能够并发处理消息的前提。
当前这个例子，数据如果乱序可以产生不同的输出结果。
数据有序SUM算子接收到的数据
数据的Eventtime按升序排列。
WarterMark推进图
每条新数据都能推进Watermark。
结果输出
所有数据都被处理，没有数据被丢弃。
数据乱序SUM算子接收到的数据
第四条事件延时到来。
WarterMark推进图
延迟的数据不会推进WaterMark，且被丢弃。
输出结果
没有统计因延迟被丢弃的第四条事件。
JOIN
将通过此例子来阐述Streaming的Retraction语义。
例子Source
SQL
广告的展现LEFT JOIN 广告的点击来更新状态：showed 或 clicked。
Sink
LEFT JOIN 可能会发送多条数据到下游。
因此必须转为RetractionStream，让下游算子有机会能撤销前次输出，从而只产生一条最终结果。
转换Table为Stream
RetractionStream没有引入特殊变化。
转为逻辑计划(Logical Plan)
优化Logical Plan
转为物理计划（Physical Plan）
优化Stream Graph
运行结果
结果数据的首个字段为标志位，True为正常数据，False为Retract数据。
RetractJoin的执行逻辑请见：NonWindowOuterJoin
ImpressionId = 1这条数据的ReactJoin执行过程。
1: Left流的Show消息先到: Show("1", "show", "2018-10-10 10:10:10")
因为之前没有输出，所以无需Retrcact。
只输出: (true, 1,2018-10-10 10:10:10,showed)
2: Right流的Click消息后到：Click("1", "click", "2018-10-10 10:13:11")
因为之前已输出过结果，所以需要Retract，输出:
(false, 1,2018-10-10 10:10:10,showed)
然后再输出新结果，
(true, 1,2018-10-10 10:10:10,clicked)
如上可知，Retraction流相当于把一条UPDATE消息分别拆成一条DELETE和一条INSERT消息。
Retraction Stream
虽然Retraction机制最多增加一倍的数据传输量，但能降低下游算子的存储负担和撤销实现难度。
传递
我们在Left Join的输出流后加一个GROUP BY，以观察Retraction流的后续算子的输出。
可能得到以下的GROUP BY输出:
由此可见，Retraction具有传递性，RetractStream的后续的Stream也会是RetractionStream。
终止
最终需要支持Retraction的Sink来终止RetractionStream，比如：
最终输出retractedResults：
存储
只有外部存储支持UPDATE或DELETE操作时，才能实现RetractionSink。
常见的KV存储和数据库，如HBase，Mysql都可实现RetractionSink。
后续程序总能从这些存储中读取最新数据，上游是否是Retraction流对用户是透明的。
常见的消息队列，如Kafka，只支持APPEND操作，则不能实现RetractionSink。
后续程序从这些消息队列可能会读到重复数据，因此用户需要在后续程序中处理重复数据。
总结
Flink Streaming SQL的实现从上到下共有三层：
其中“Streaming Data Model” 和 “Distributed Snapshot” 是Flink这个分布式流计算系统的核心架构设计。
“Streaming Data Model”的What, Where, When, How 明确了流计算系统的表达能力及预期应用场景。
“Distributed Snapshots”针对预期的应用场景在数据准确性，系统稳定性和运行性能上做了合适的折中。
本文通过实例阐述了流计算开发者需要了解的最上面两层的概念和原理，
以便流计算开发者能在数据准确性和数据时效性上做适合业务场景的折中和取舍。
