Flink Table/SQL API 规划 —— Dynamic Table

动态表的概念是社区很早就提出的但并无所有实现，下文中全部介绍都是基于已有规划和proposal给出的，可能与以后实现存在出入仅供参考复制代码

概念

动态表直观上看是一个相似于数据库中的Materialized View概念。动态表随着时间改变；相似静态的batch table同样能够用标准SQL进行查询而后一个新的动态表；能够和流无损地互相转换(对偶的)。对现有的API最大的改进关键在表的内容随着时间改变，而如今的状态只是append。当前的streaming table能够认为是一种动态表，append模式的动态表。

流到 Dynamic Table

流被转换成Table时决定选择哪一种模式是依据表的schema是否认义primary key。

Append模式：

若是表的schema没有包括key的定义那转换成表时采用append模式。把流中每条新来的record当作新的row append到表中。一旦数据加到表中就不能再被更新和删除(指当前表中，不考虑转换成新表)。

Replace模式：

相对应，若是定义了key，那么对于流中的每条记录若是key不在表中就insert不然就update。

Dynamic Table 到 流

表到流的操做是把表的全部change以changelog stream的方式发送到下游。这一步也有两种模式。

Retraction模式：

traction模式中对于Dynamic Table的insert和delete的change分别产生insert或delete event。若是是update的change会产生两种change event，对于以前发送出去的一样key的record会产生delete event，对于当前的record是产生insert event。以下图所示：

Update模式：

update模式依赖Dynamic Table定义了key。全部的change event是一个kv对。key对应表的key在当前record中的值；对于insert和change value对应新的record。对于delete value是空表示该能够已经被删除。以下图所示：

example

表的内容随着时间改变意味着对表的query结果也是随着时间改变的。咱们定义：

• A[t]: 时间t时的表A
• q(A[t])：时间t时对表A执行query q

举个例子来理解动态表的概念：

query的限制

因为流是无限的，相对应 Dynamic Table 也是无界的。当查询无限的表的时候咱们须要保证query的定时是良好的，有意义可行的。

1.在实践中Flink将查询转换成持续的流式应用，执行的query仅针对当前的逻辑时间，因此不支持对于任意时间点的查询(A[t])。
2.最直观的原则是query可能的状态和计算必须是有界的，因此能够支持可增量计算的查询：

• 不断更新当前结果的查询：查询能够产生insert，update和delete更改。查询能够表示为 Q(t+1) = q'(Q(t), c(T, t, t+1))，其中Q(t)是query q的前一次查询结果，c(T, t, t_+1) 是表T从t+1到t的变化, q'是q的增量版本。
• 产生append-only的表，能够从输入表的尾端直接计算出新数据。查询能够表示为 Q(t+1) = q''(c(T, t-x, t+1)) ∪ Q(t)，q''是不须要时间t时q的结果增量版本query q。c(T, t-x, t+1)是表T尾部的x+1个数据，x取决于语义。例如最后一小时的window aggregation至少须要最后一小时的数据做为状态。其余能支持的查询类型还有：单独在每一行上操做的SELECT WHERE；rowtime上的GROUP BY子句（好比基于时间的window aggregate）；ORDER BY rowtime的OVER windows(row-windows)；ORDER BY rowtime。
  3.当输入表足够小时，对表的每条数据进行访问。好比对两个大小固定的流表(好比key的个数固定)进行join。

中间状态有界

如上文所说的，某些增量查询须要保留一些数据（部分输入数据或者中间结果）做为状态。为了保证query不会失败，保证查询所须要的空间是有界的不随着时间无限增加很重要。主要有两个缘由使得状态增加：

1. 不受时间谓词约束的中间计算状态的增加（好比 聚合key的膨胀）
2. 时间有界可是须要迟到的数据（好比 window 的聚合）

虽然第二种状况可有经过下文提到的"Last Result Offset"参数解决，可是第一种状况须要优化器检测。咱们应该拒毫不受时间限制的中间状态增加的查询。优化器应该提供如何修复查询且要求有适当的时间谓词。好比下面这个查询：

SELECT user, page, COUNT(page) AS pCnt
FROM pageviews
GROUP BY user, page复制代码

随着用户数和页面数的增加，中间状态会数据随着时间推移而增加。对于存储空间的要求能够经过添加时间谓词来限制：

SELECT user, page, COUNT(page) AS pCnt
FROM pageviews
WHERE rowtime BETWEEN now() - INTERVAL '1' HOUR AND now() // only last hour
GROUP BY user, page复制代码

由于不是全部属性都是不断增加的, 所以能够告诉优化器domain的size, 就能够推断中间状态不会随着时间推移而增加，而后接受没有时间谓词的查询。

val sensorT: Table = sensors
  .toTable('id, 'loc, 'stime, 'temp)
  .attributeDomain('loc, Domain.constant) // domain of 'loc is not growing 
env.registerTable("sensors", sensorT)

SELECT loc, AVG(temp) AS avgTemp
FROM sensors
GROUP BY loc复制代码

结果的计算和细化时序

一些关系运算符必须等数据到达才能计算最终结果。例如：在10：30关闭的窗口至少要等到10：30才能计算出最终的结果。Flink的logical clock（即 决定什么时候才是10：30）取决于使用event time 仍是 processing time。在processing time的状况下，logical time是每一个机器的wallclock；在event time的状况下，logical clock time是由源头提供的watermark决定的。因为数据的乱序和延迟当在event time模式下时等待一段时间来减少计算结果不完整性。另外一方面某些状况下但愿获得不断改进的早期结果。所以对于结果被计算、改进或者作出最终结果时有不一样的要求、

下图描绘了不一样的配置参数如何用于控制早期结果和细化计算结果的。

• "First Result Offset" 指第一个早期结果被计算的结果的时间。时间是相对于第一次能够计算完整结果的时间（好比相对于window的结束时间10:30）。若是设置的是-10分钟，对于结束时间是10：30的window那么第一个被发出去的结果是在逻辑时间10：20计算的。这个参数的默认值是0，即在window结束的时候才计算结果。
• "Complete Result Offset" 指完整的结果被计算的时间。时间是相对于第一次能够计算完整的时间。若是设置的是+5分钟，对于结束时间是10：30的window那么产生完整结果的时间是10：35。这个参数能够减轻延迟数据形成的影响。默认是0，即在window结束的时候计算的结果就是完整结果。
• "Update Rate" 指计算完整结果以前一次次更新结果的时间间隔（能够是时间和次数）。若是设为5分钟，窗口大小是30分钟的tumbling window，开始时间是10：300，"First Result Offset"是-15分钟， "Complete Result Offset"是2分钟，那么将在10:20, 10:25, 10：30更新结果，10:15禅城寄一个结果，10：32产生完整结果。
• "Last Updates Switch" 指完整结果发出后对于延迟的数据是否计算延迟更新，直到计算状态被清除。
• "Last Result Offset" 指可计算的最后一个结果的时间。这是内部状态被清除的时间，清除状态后再到达的数据将被丢弃。Last Result Offset 意味着计算的结果是近似值，不能保证精确。