动态表

数据流上的关系查询

下表比较了传统的关系型数据库和流处理在输入数据、执行和输出结果方面的不同。

SQL	流处理
关系（或表）是有界（多）元组的集合。	流是无限的元组序列。
对批处理数据执行的查询（例如，关系数据库中的表）可以访问完整的输入数据。	流式查询在启动时无法访问所有数据，必须“等待”数据流入。
批处理查询在生成固定大小的结果后终止。	流式查询会根据收到的记录不断更新其结果，并且永远不会完成。

动态表与连续查询

动态表是Flink的Table API和SQL支持流数据的核心概念。 与表示批处理数据的静态表相比，动态表随时间而变化，可以像静态批处理表一样查询它们。 查询动态表会产生连续查询。 连续查询永远不会终止并生成动态表作为结果。 查询不断更新其（动态）结果表以反映其（动态）输入表的更改。

下图展示了流，动态表和连续查询的关系：

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

在流上定义表

为了使用关系查询处理流，必须将其转换为表。从概念上讲，流的每个记录都被解释为结果表上的插入修改。

连续查询

在动态表上对连续查询进行计算，并生成新的动态表。与批处理查询不同，连续查询从不根据输入表上的更新来终止和更新其结果表。在任何时间点上，连续查询的结果在语义上都等同于在输入表的快照上以批处理模式执行的相同查询的结果。

查询限制

许多语义有效的查询都可以作为流上的连续查询进行计算。有些查询的计算开销太大，要么是因为它们需要维护的状态的大小，要么是因为计算更新的开销太大。

• 状态大小:连续查询是在无界流上的计算，通常需要运行数周或数月。因此，连续查询过程的数据总量可能非常大。必须更新以前发出的结果的查询需要维护所有发出的行，以便能够更新它们。例如，第一个示例查询需要存储每个用户的URL计数，以便在输入表接收到新行时能够增加计数并发送新的结果。如果只跟踪注册用户，则维护的计数可能不会太高。但是，如果非注册用户获得分配的唯一用户名，则要维护的计数数量将随着时间的推移而增加，并可能最终导致查询失败。
• 计算更新:即使只添加或更新了单个输入记录，某些查询也需要重新计算和更新大部分发出的结果行。显然，这样的查询不太适合作为连续查询执行。下面的查询是一个示例，它根据最后一次单击的时间为每个用户计算一个等级。一旦clinks表接收到新行，就会更新用户的lastAction，并且必须计算新的排名。然而，由于两行不能具有相同的等级，因此所有排名较低的行也需要更新。

表到流的转换

动态表可以像常规数据库表一样，通过插入、更新和删除来不断修改。它可能是一个只有一行的表(不断更新)，一个只有插入的表(没有更新和删除修改)，或者介于两者之间。
将动态表转换为流或将其写入外部系统时，需要对这些更改进行编码。Flink的Table API和SQL支持三种方法来编码动态表的更改：

• 仅追加流(Append-only stream): 仅通过Insert操作修改的动态表可以通过发出插入的行转换为流。
• 回收流(Retract stream):回收流是具有两种类型的消息的流，添加消息和回收消息。 通过将INSERT更改编码为添加消息，将DELETE更改编码为回收消息，将UPDATE更改编码为更新（先前）行的回收消息和更新（新）行的添加消息，将动态表转换为回收流。
• 更新插入流(Upsert stream):upsert流是一种包含两种类型消息的流，包括Upsert消息和删除消息。 转换为upsert流的动态表需要（可能是复合的）唯一键。 通过将INSERT和UPDATE更改编码为upsert消息并将DELETE更改编码为删除消息，将具有唯一键的动态表转换为流。 流消耗运算符需要知道唯一键属性才能正确应用消息。 与回收流的主要区别在于UPDATE更改使用单个消息进行编码，因此更有效。

时态表

时态表表示历史表上的(参数化)视图的概念，它在特定的时间点返回表的内容。在Flink中，时态表由时态表函数表示。

时态表函数

为了访问时态表中的数据，必须传递一个时间属性，该属性确定将返回的表的版本。Flink使用表函数的SQL语法提供了一种表示它的方法。
一旦定义，时态表函数接受单个时间参数timeAttribute并返回一组行。该集合包含与给定时间属性相关的所有现有主键的最新行版本。

时间属性

Flink能够根据不同的时间概念处理流数据。

• 处理时间(Processing Time)：指执行相应操作的机器的系统时间(也称为“挂钟时间”)。
• 摄取时间(Ingestion Time )：事件进入Flink的时间;在内部，它的处理类似于事件时间。
• 事件时间(Event Time)：指基于附加到每行上的时间戳处理流数据的过程。时间戳可以在事件发生时进行编码。

时间属性介绍

Table API和SQL中基于时间的操作（如窗口）需要有关时间概念及其来源的信息。因此，表可以提供逻辑时间属性，用于指定时间和访问表程序中的相应时间戳。
时间属性可以是每个表模式的一部分。它们是在从DataStream创建表时定义的，或者是在使用表源时预定义的。一旦开始定义了时间属性，就可以将其引用为字段，并在基于时间的操作中使用。

val env = StreamExecutionEnvironment.getExecutionEnvironment

env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime) // default

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

持续查询中的关联

常规关联

常规关联是最通用的关联类型，在这种关联类型中，关联输入的任何一边的任何新记录或更改都是可见的，并且会影响整个连接结果。例如，如果左侧有一条新记录，它将与右侧的所有以前和将来的记录关联。

SELECT * FROM Orders
INNER JOIN Product
ON Orders.productId = Product.id

这些语义允许任何类型的插入、更新、删除输入表。但是，这个操作有一个重要的含义:它要求关联输入的两侧永久保持在Flink的状态中。因此，如果一个或两个输入表都在持续增长，那么资源使用也会无限增长。

时间窗口关联

时间窗口连接由关联谓词定义，关联谓词检查输入记录的时间属性是否在某些时间限制内，即时间窗口。

SELECT *
FROM
  Orders o,
  Shipments s
WHERE o.id = s.orderId AND
      o.ordertime BETWEEN s.shiptime - INTERVAL '4' HOUR AND s.shiptime

与常规关联操作相比，这种关联只支持具有时间属性的仅追加表。由于时间属性是准一元递增的，Flink可以在不影响结果正确性的情况下从其状态中删除旧值。

时态表关联

与时态表的关联将仅追加表(左输入/探测端)与时态表(右输入/构建端)关联起来，即，一个随时间变化并跟踪其变化的表。

处理时间（Processing-time）时态关联

使用处理时间属性，不可能将过去的时间属性作为参数传递给时态表函数。根据定义，它始终是当前时间戳。因此，处理时间时态表函数的调用将始终返回基础表的最新已知版本，并且基础历史表中的任何更新也将立即覆盖当前值。

事件时间（Event-time）时态关联

利用事件时间属性（即行时属性），可以将过去的时间属性传递给时态表函数。这允许在共同的时间点关联两个表。