如何在Flink SQL批处理模式下查询Kafka Upsert表？

问题背景 #

你通过以下Flink SQL创建了Upsert-Kafka表：

CREATE TABLE en_trans (
  `transid` INTEGER,
  `productname` INTEGER,
   PRIMARY KEY (transid) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'en_trans',
  'properties.bootstrap.servers' = '....:9092',
  'properties.group.id' = 'en_trans_group_test',
  'key.format' = 'avro-confluent',
  'value.format' = 'avro-confluent',
  'key.avro-confluent.url' = 'http://kafka-netlex-cp-schema-registry:8081',
  'value.avro-confluent.url' = 'http://kafka-netlex-cp-schema-registry:8081'
);

执行批模式查询时：

select * from en_trans where transid=123;

触发错误：

org.apache.flink.table.api.ValidationException: Querying an unbounded table 'default_catalog.default_database.en_trans' in batch mode is not allowed. The table source is unbounded.

由于需要提供带分页的REST服务，不想使用流处理模式，可通过以下方案解决：

解决方案 #

1. 同步Kafka数据到批存储系统

将Upsert-Kafka中的数据同步到支持批查询的存储（如MySQL、Hive、PostgreSQL等），之后直接在批模式下查询该存储的表，天然支持分页和点查。

示例（同步到MySQL）：
首先创建MySQL的Sink表：

CREATE TABLE en_trans_mysql (
  `transid` INTEGER PRIMARY KEY NOT ENFORCED,
  `productname` INTEGER
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://your-mysql-host:3306/your-db',
  'table-name' = 'en_trans',
  'username' = 'your-username',
  'password' = 'your-password'
);

然后启动流任务将Kafka表的数据同步到MySQL：

INSERT INTO en_trans_mysql SELECT * FROM en_trans;

之后即可在批模式下执行带分页的查询：

-- 点查
SELECT * FROM en_trans_mysql WHERE transid = 123;
-- 分页查询
SELECT * FROM en_trans_mysql LIMIT 10 OFFSET 20;

2. 使用物化视图维护全量快照

通过Flink的物化视图持续消费Kafka数据并维护全量状态，之后在流模式下查询物化视图，可获得类似批查询的体验，支持点查和分页。

创建物化视图：

CREATE MATERIALIZED VIEW en_trans_mv AS
SELECT transid, productname FROM en_trans;

查询物化视图：

-- 点查
SELECT * FROM en_trans_mv WHERE transid = 123;
-- 分页查询
SELECT * FROM en_trans_mv LIMIT 10 OFFSET 20;

注意：需为Flink配置合适的状态后端和检查点，确保物化视图的数据一致性和可恢复性。

3. 启用流批一体模式（快速验证）

Flink支持流批一体，可通过配置让查询以流模式执行，但返回当前的全量快照结果。需将执行模式设置为STREAMING，之后直接执行查询：

SET execution.runtime-mode = STREAMING;

-- 点查
SELECT * FROM en_trans WHERE transid = 123;
-- 分页查询
SELECT * FROM en_trans LIMIT 10 OFFSET 20;

这种方式本质仍是流处理，但能获取Kafka当前的全量数据快照，适合快速验证场景；但需注意状态大小和数据实时性的平衡。

内容的提问来源于stack exchange，提问作者taymedee