1.概述

转载：https://blog.csdn.net/u013939918/article/details/107623778

流计算的一个典型场景是把聚合的数据写入到 Upsert Sink 中，比如 JDBC、HBase，当遇到复杂的 SQL 时，时常会出现：

UpsertStreamTableSink 需要上游的 Query 有完整的 Primary Key 信息，不然就直接抛异常。这个现象涉及到 Flink 的 UpsertStreamTableSink 机制。顾名思义，它是一个更新的 Sink，需要按 Key 来更新，所以必须要有 Key 信息。

如何发现 Primary Key？一个方法是让优化器从 Query 中推断，如下图发现 Primary Key 的例子。

这种情况下在简单 Query 当中很好，也满足语义，也非常自然。但是如果是一个复杂的 Query，比如聚合又 Join 再聚合，那就只有报错了。不能期待优化器有多智能，很多情况它都不能推断出 PK，而且，可能业务的 SQL 本身就不能推断出 PK，所以导致了这样的异常。

怎么解决问题？Flink 1.11 彻底的抛弃了这个机制，不再从 Query 来推断 PK 了，而是完全依赖 Create table 语法。比如 Create 一个 jdbc_table，需要在定义中显式地写好 Primary Key（后面 NOT ENFORCED 的意思是不强校验，因为 Connector 也许没有具备 PK 的强校验的能力）。当指定了 PK，就相当于就告诉框架这个Jdbc Sink 会按照对应的 Key 来进行更新。如此，就跟 Query 完全没有关系了，这样的设计可以定义得非常清晰，如何更新完全按照设置的定义来。

CREATE TABLE jdbc_table (
    id BIGINT,
    ...
    PRIMARY KEY (id) NOT ENFORCED
)