Flink 的 Table API 和SQL支持是用于批处理和流处理的统一API。这意味着 Table API 和SQL查询具有相同的语义，无论它们的输入是有界批量输入还是无界流输入。因为关系代数（relational algebra）和SQL最初是为批处理而设计的，所以对于无界流输入的关系查询不像有界批输入上的关系查询那样容易理解。下面将解释 Flink 关于流数据的关系API的概念、实际限制和特定于流的配置参数。

流数据上的关系查询

SQL和 Relational algebra 并没有考虑到流数据。因此，在关系代数（和SQL）和流处理之间有一些概念上的差距。

尽管存在这些差异，但使用关系查询和SQL处理流数据并非不可能。高级关系数据库系统提供称为物化视图（Materialized View）的特性。物化视图定义为SQL查询，就像常规虚拟视图一样。与虚拟视图相比，物化视图缓存查询的结果，使得在访问视图时不需要评估查询。缓存的一个常见挑战是防止缓存提供过时的结果。当修改其查询的基表时，物化视图将过时。Eager View Maintenance 是一种更新物化视图和在基表更新后立即更新物化视图的技术。

如果我们考虑以下因素，那么在流上的 Eager view maintenance 和SQL查询之间的联系就会变得明显：

• 数据库表是应用的 INSERT，UPDATE 和 DELETE 命令的流数据的结果流，通常被称为 changelog stream。
• 物化视图定义为SQL查询。为了更新视图，查询不断地处理视图关联的 changelog stream。
• 物化视图是流式SQL查询的结果。

动态表和连续查询

动态表（Dynamic table）是 Flink Table API 和SQL支持流数据的核心概念。与表示批处理数据的静态表（static table）相比，动态表会随时间而变化，并且可以像静态批处理表一样查询。查询动态表会生成连续查询（Continuous Query）。连续查询永远不会终止并生成动态表作为结果。查询不断更新其结果表以反映其输入表的更改。实质上，对动态表的连续查询与物化视图的查询非常相似。

连续查询的结果在语义上总是等价于在输入表的快照上以批处理模式执行的相同查询的结果。下图显示了流、动态表和连续查询的关系：

1. 流转换为动态表
2. 在动态表上连续查询，生成新的动态表
3. 生成的动态表将转换回流

Stream query

动态表首先是一个逻辑概念。在查询执行期间，不必（完全）实现动态表。

在下文中，我们将用具有以下模式的点击事件（click events）的流解释动态表和连续查询的概念：

[ 
  user:  VARCHAR,   // the name of the user
  cTime: TIMESTAMP, // the time when the URL was accessed
  url:   VARCHAR    // the URL that was accessed by the user
]

在流上定义表

为了使用关系查询处理流，必须将其转换为表。从概念上讲，流的每个记录都被解释为对结果表的 INSERT 修改。下图显示了点击事件流（左侧）如何转换为表（右侧）。随着更多的点击事件的插入，结果表不断增长。

Append mode

连续查询

在动态表上进行连续查询，并生成新的动态表。与批查询相反，连续查询不会停止更新其结果表。在任何时间点，连续查询的结果在语义上等同于在输入表的快照上以批处理模式执行的相同查询的结果。

在下面展示了在点击事件流中定义的 clicks 表上的两个查询的例子。

第一个查询是一个简单的 GROUP-BY COUNT 聚合查询。在 clicks 表上按 user 字段进行分组，并计算访问的URL数量。下图显示了随着 clicks 表的行数增加更新查询：

1. 当查询启动时，clicks 表（左侧）为空。第一行记录 insert clicks 表时，开始计算结果表。
2. clicks 插入第一行 [Mary, ./home]，结果表（右侧，顶部）为 [Mary, 1]。
3. clicks 插入第二行 [Bob, ./cart]，结果表插入新行 [Bob, 1]。
4. clicks 插入第三行 [Mary, ./prod?id=1]，更新结果表，[Mary, 1] 更新为 [Mary, 2]。
5. clicks 插入第四行 [Liz, 1]，结果表插入新行 [Liz, 1]。。

Group by & Count

第二个查询类似于第一个查询，但 clicks 表除了 user 字段之外新增了 cTime 字段，并且按小时生成滚动窗口，然后计算URL数量。同样，下图显示了不同时间点的输入和输出，以显示动态表的变化：

1. 查询每小时连续计算结果并更新结果表。
2. 第一个窗口（12:00:00 ~ 12:59:59），clicks 表包含四行记录输入，查询计算得到两个结果行并加入到结果表。
3. 对于下一个窗口（13:00:00 ~ 13:59:59），clicks 表包含三行记录输入，这导致新增两行结果被追加到结果表中。
4. 随着时间的推移，更多的行被追加到 clicks 表中，结果表将被更新。

Group by Window Count

尽管两个查询例子看起来非常相似（都计算了分组计数聚合），但在一个重要方面有所不同：

• 第一个查询更新之前发出的结果，即结果表的 changelog 包含 INSERT 和 UPDATE。
• 第二个查询仅附加到结果表，即结果表的 changelog 仅包含 INSERT。

查询限制

许多（但不是全部）语义有效的查询可以作为流上的连续查询进行计算。有些查询的计算成本太高，要么是因为它们需要维护的状态太大，要么是因为计算更新太昂贵：

• 状态大小（State Size）：连续查询在无界流上进行计算，连续查询处理的数据总量可能非常大。必须更新先前发出的结果，因此需要维护所有已经发出的行。例如，第一个示例需要存储每个 user 的URL计数，以便在输入表收到新行时能够更新计数。

SELECT user, COUNT(url)
FROM clicks
GROUP BY user;

• 计算更新（Computing Updates）：某些查询，即使只添加或更新了单个输入记录，也需要重新计算和更新大部分发出的结果行。显然，这样的查询不适合作为连续查询执行。例如以下查询，该查询基于最后一次点击的时间排序每个用户。一旦 clicks 表收到新行或某一行 lastLogin 更新，并且必须计算新的排名。

SELECT user, RANK() OVER (ORDER BY lastLogin)
FROM (
  SELECT user, MAX(cTime) AS lastAction FROM clicks GROUP BY user
);

表转换到流

动态表可以像常规数据库表一样，通过 INSERT、UPDATE 和 DELETE 更改来不断修改。

将动态表转换为流或将其写入外部系统时，需要对这些更改进行编码。Flink Table API 和SQL支持三种动态表更改的方法：

• Append-only stream：只能由插入修改的动态表，可以通过发出插入的行转换为流。

• Retract stream：Retract stream 是具有两种类型的消息的流，添加消息（add）和撤消消息（retract）。通过将插入编码为添加消息（add message），将删除编码为撤销消息（retract message），将更新编码为撤销消息（更新先前行）和添加消息（更新新行），将动态表转换为 Retract stream。

Undo redo

• Upsert stream： Upsert stream 是具有两种类型的消息的流，更新插入消息（upsert）和删除消息（delete）。转换为 Upsert stream 的动态表需要有唯一键（可能是复合的），插入和更新编码为 upsert 消息，删除编码为 delete 消息。算子需要知道唯一键属性才能正确消费消息，与 Retract stream 的区别在于：更新修改使用单个消息进行编码，更加有效。下图显示了动态表到 Upsert stream 的转换。

Redo

时间属性

Flink 可以根据不同的时间域处理流数据：

1. Processing Time：执行各个算子操作的系统时间
2. Event Time：事件发生时间，附加在每条记录上的时间戳
3. Ingestion Time：事件进入 Flink 的时间，与事件时间类似
   更多关于Flink时间处理的信息，请参考 Event Time 和 Watermarks

val env = StreamExecutionEnvironment.getExecutionEnvironment

// default
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime) 

// alternatively:
// env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
// env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

Table API 和SQL查询中基于时间的算子操作（如 Window）需要设置时间域及时间提取信息。因此，表可以提供逻辑时间属性，用于指示时间和访问表程序中相应的时间戳。

时间属性可以是每个表模式中的一部分，可以从 DataStream 中创建表时创建，也可以在使用 TableSource 时预定义。一旦开始定义了时间属性，它就可以作为字段引用，并可以用于基于时间的 算子操作。

只要时间属性未被修改，并且只是从查询的一部分转发到另一部分，那么它仍是一个有效的时间属性。时间属性跟常规时间戳一样，可以用于计算，一旦被用于计算，它会具体化并成为一个常规时间戳。常规时间戳跟 Flink 的时间和水印系统没有关系，所以不会被用来做基于时间的算子操作。

Processing Time

处理时间（Processing Time）允许程序根据本地时间产生结果，这是最简单的时间概念，既不要求时间戳提取也不会产生水印。有两种方法可以定义处理时间属性。

DataStream 到 Table 的转换中定义

处理时间属性是在 Schema 定义时使用 .proctime 属性来定义的。时间属性只能通过在 Schema 的基础上新增一个逻辑字段来扩展，因此，它只能在模式定义的末尾来定义。

val stream: DataStream[(String, String)] = ...

// 声明一个逻辑字段，作为 processing time 属性
val table = tEnv.fromDataStream(stream, 'UserActionTimestamp, 'Username, 'Data, 'UserActionTime.proctime)

val windowedTable = table.window(Tumble over 10.minutes on 'UserActionTime as 'userActionWindow)

TableSource 定义

处理时间属性可以通过一个实现了 DefinedProctimeAttribute 接口的 TableSource 定义，逻辑时间属性会被添加到由 TableSource 的返回类型定义的 Schema 中。

// 定义 Table source，并指定 processing time 属性
class UserActionSource extends StreamTableSource[Row] with DefinedProctimeAttribute {

    override def getReturnType = {
        val names = Array[String]("Username" , "Data")
        val types = Array[TypeInformation[_]](Types.STRING, Types.STRING)
        Types.ROW(names, types)
    }

    override def getDataStream(execEnv: StreamExecutionEnvironment): DataStream[Row] = {
        // create stream
        val stream = ...
        stream
    }

    override def getProctimeAttribute = {
        // field with this name will be appended as a third field 
        "UserActionTime"
    }
}

// register table source
tEnv.registerTableSource("UserActions", new UserActionSource)

val windowedTable = tEnv
    .scan("UserActions")
    .window(Tumble over 10.minutes on 'UserActionTime as 'userActionWindow)

Event Time

事件时间允许程序根据包含在每条记录中的时间生成结果。即使对于无序或者延迟事件也会产生一致性结果。当从持久存储中读取记录时，它还确保表程序的可重放结果。

此外，事件时间允许在批环境和流环境中使用统一的语法，流式环境中的时间属性可以是批处理环境中的记录的常规字段。

为了处理流事件无序，并区分准时和延迟事件，Flink 需要抽取事件中的时间戳，并且在生成水印来描述进展。有两种方法可以定义事件时间属性。

DataStream 到 Table 的转换中定义

事件时间可以在 shcema 定义中通过 .rowtime 属性来定义。Timestamp 和 watermark 必须在 DataStream 的转换中就已经指定好。有两种方式来定义时间属性:

1. 添加一个新的字段到 Schema
2. 替换一个现有的字段

无论哪种方式，事件时间字段都将保存数据流事件时间戳的值。

// Option 1:

// 提取时间戳并指定水印 
val stream: DataStream[(String, String)] = inputStream.assignTimestampsAndWatermarks(...)

// 定义一个额外的逻辑字段作为 event time
val table = tEnv.fromDataStream(stream, 'Username, 'Data, 'UserActionTime.rowtime)


// Option 2:

// 提取时间戳并指定水印 
val stream: DataStream[(Long, String, String)] = inputStream.assignTimestampsAndWatermarks(...)

// 第一个字段被用作时间戳提取，不需要定义额外的字段
val table = tEnv.fromDataStream(stream, 'UserActionTime.rowtime, 'Username, 'Data)

// Usage:

val windowedTable = table.window(Tumble over 10.minutes on 'UserActionTime as 'userActionWindow)

TableSource 定义

处理时间属性可以通过一个实现了 DefinedRowtimeAttribute 接口的 TableSource 定义。getRowtimeAttribute() 方法返回一个现有字段的名称，该字段包含表的事件时间属性，并且是类型 LONG 或 TIMESTAMP。

此外，getDataStream() 方法返回的 DataStream 必须分配与定义的时间属性对齐的水印。

// 定义 Table source，并指定 rowtime 属性
class UserActionSource extends StreamTableSource[Row] with DefinedRowtimeAttribute {

    override def getReturnType = {
        val names = Array[String]("Username" , "Data", "UserActionTime")
        val types = Array[TypeInformation[_]](Types.STRING, Types.STRING, Types.LONG)
        Types.ROW(names, types)
    }

    override def getDataStream(execEnv: StreamExecutionEnvironment): DataStream[Row] = {
        // create stream
        // ...
        // assign watermarks based on the "UserActionTime" attribute
        val stream = inputStream.assignTimestampsAndWatermarks(...)
        stream
    }

    override def getRowtimeAttribute = {
        // Mark the "UserActionTime" attribute as event-time attribute.
        "UserActionTime"
    }
}

// register the table source
tEnv.registerTableSource("UserActions", new UserActionSource)

val windowedTable = tEnv
    .scan("UserActions")
    .window(Tumble over 10.minutes on 'UserActionTime as 'userActionWindow)

查询配置

Flink的Table API和SQL接口使用QueryConfig来控制计算、发射的结果以及更新发射结果。

Table API 和SQL查询具有相同的语义，无论它们的输入是有界批量输入还是无界流输入。在许多情况下，对流输入的连续查询能够与离线计算结果有相同的准确结果。然而，这在一般情况下是不可能的，因为连续查询必须限制它们维护的状态的大小，以避免耗尽存储空间，并且能够长时间处理无界流数据。所以，连续查询可能只能提供近似结果，具体取决于输入数据和查询本身。

Flink Table API 和SQL接口提供参数来调整连续查询的准确性和资源消耗。参数通过 QueryConfig 对象指定。QueryConfig 可以从 TableEnvironment 获得。

val env = StreamExecutionEnvironment.getExecutionEnvironment
val tableEnv = TableEnvironment.getTableEnvironment(env)

// 获取 query configuration
val qConfig: StreamQueryConfig = tableEnv.queryConfig

// 设置查询参数
qConfig.withIdleStateRetentionTime(Time.hours(12), Time.hours(24))

// 定义查询和 TableSink
val result: Table = ???
val sink: TableSink[Row] = ???

// TableSink 发送结果表时传递查询参数
result.writeToSink(sink, qConfig)

// 转换为 DataStream 时传递查询参数
val stream: DataStream[Row] = result.toAppendStream[Row](qConfig)

空闲状态保存时间

许多查询在一个或多个属性上进行聚合或连接操作。当在流上执行此类查询时，连续查询需要收集记录或维护每个键的结果值。如果输入流的关键域正在演变，即活跃的键值随时间变化，观察到越来越多的不同键，连续查询会累积越来越多的状态。但是，通常在一段时间后一些键变为非活跃，其对应的状态变得陈旧且无用。

例如，以下查询计算每个会话的单击次数。

SELECT sessionId, COUNT(*) FROM clicks GROUP BY sessionId;

sessionId 属性用于分组，连续查询维护每个 sessionId 及其计数。sessionId 属性随着时间的推移而发展，并且 sessionId 值仅在会话结束之前有效（在有限的时间段内）。但是，连续查询无法知道 sessionId 有此属性，并期望每个 sessionId 值都可以在任何时间点发生。因此会维护每个观察到的 sessionId 的计数。随着 sessionId 观察到越来越多，查询的总状态大小不断增长。

空闲状态保持时间（Idle State Retention Time）参数定义一个键的状态在一次更新之后保存多久后删除。对于前面的查询示例，sessionId 只要在配置的时间段内没有更新，就会删除对应计数。

通过删除键的状态，连续查询会完全忘记它之前已经看过这个键。如果删除的键再次出现，则被视为具有相应键的第一个记录。对于前面的查询示例，这意味着 sessionId 的计数从0开始。

配置空闲状态保存时间有两个参数：

• minimum idle state retention time，定义非活动键的状态在删除前至少保持多少时间。
• maximum idle state retention time，定义非活动键的状态在删除前最多保持多少时间。

对于前面的查询示例：

val qConfig: StreamQueryConfig = ???

// 设置 idle state retention time: min = 12 hours, max = 24 hours
qConfig.withIdleStateRetentionTime(Time.hours(12), Time.hours(24))

清理状态需要额外的记录，对于 minTime 和 maxTime 较大差异的情况成本更低，因此 minTime 和 maxTime 直接必须至少相差5分钟。

Reference：
https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/table/streaming.html