我已经在 Flink 文档和官方 Flink 博客中看到过多次提到基于唯一键的动态表的“upsert 模式”。但是，我没有看到任何关于如何在动态表上启用此模式的示例/文档。
例子：
当通过更新模式在流上定义动态表时，我们可以在表上指定一个唯一键属性。在这种情况下，对键属性执行更新和删除操作。 更新模式如下图所示。
转换为 upsert 流 的动态表需要一个（可能是复合的）唯一键。
所以我的问题是：
【问题讨论】：
标签： apache-flink flink-streaming flink-sql
链接的资源描述了两种不同的场景。
DataStream -> Table 转换。 Table -> DataStream 转换。以下讨论基于 Flink 1.4.0（2018 年 1 月）。
Upsert DataStream -> Table 转化
本机不支持通过键上的 upsert 将 DataStream 转换为 Table，但在路线图上。同时，您可以使用附加 Table 和具有用户定义聚合函数的查询来模拟此行为。
如果您有一个追加 Table Logins 与架构 (user, loginTime, ip) 跟踪用户的登录，您可以将其转换为一个 upsert Table 键入 user 使用以下查询：
SELECT user, LAST_VAL(loginTime), LAST_VAL(ip) FROM Logins GROUP BY user
LAST_VAL 聚合函数是一个user-defined aggregation function，它始终返回最新的附加值。
对 upsert DataStream -> Table 转换的本机支持基本上以相同的方式工作，但提供了更简洁的 API。
Upsert Table -> DataStream 转化
不支持将 Table 转换为 upsert DataStream。这也正确反映在文档中：
请注意，将动态表转换为 DataStream 时，仅支持追加和收回流。
我们故意选择不支持 upsert Table -> DataStream 转换，因为只有知道关键属性时才能处理 upsert DataStream。这些取决于查询，并不总是很容易识别。开发人员有责任确保正确解释关键属性。不这样做会导致错误的程序。为避免出现问题，我们决定不提供 upsert Table -> DataStream 转换。
相反，用户可以将Table 转换为撤回DataStream。此外，我们支持 UpsertTableSink 将 upsert DataStream 写入外部系统，例如数据库或键值存储。
【讨论】：
更新：从 Flink 1.9 开始，LAST_VALUE 是 build-in aggregate functions 的一部分，如果我们使用 Blink planner（这是 Flink 1.11 以来的默认设置）。
假设上面 Fabian Hueske 的回复中提到的 Logins 表存在，我们现在可以将其转换为 upsert 表，如下所示：
SELECT
user,
LAST_VALUE(loginTime),
LAST_VALUE(ip)
FROM Logins
GROUP BY user
【讨论】：
Flink 1.8 仍然缺乏这样的支持。预计将来会添加这些功能：1) LAST_VAL 2) Upsert Stream Dynamic Table.
ps。 LAST_VAL() 似乎不可能在 UDTF 中实现。聚合函数不提供附加的事件/过程时间上下文。阿里巴巴的 Blink 提供了 LAST_VAL 的替代实现，但它需要另一个字段来提供订单信息，而不是直接在 event/proc time 上。这使得 sql 代码很难看。 (https://help.aliyun.com/knowledge_detail/62791.html)
我的 LAST_VAL（例如获取最新 ip）的变通解决方案类似于：
【讨论】：