Flink JDBC SQL Connector 用一张 DDL 打通任意关系型数据库（Scan / 维表 Join / Upsert 落库 / Catalog）

1、能力速览：Scan、Lookup、Sink 都齐了

官方给 JDBC SQL Connector 的能力标签很明确： (nightlies.apache.org)

• Scan Source：Bounded（有界扫描，适合批读）
• Lookup Source：Sync Mode（同步维表查询，用于 temporal join）
• Sink：Batch + Streaming
• Sink 支持 Streaming Append & Upsert Mode（关键在“有没有主键”）

核心规则只有一句话：

• DDL 定义了PRIMARY KEY→ Sink 走Upsert，能承接 UPDATE/DELETE（changelog）
• DDL 没定义主键 → Sink 只能Append（INSERT-only），不支持消费 UPDATE/DELETE (nightlies.apache.org)

这也是为什么很多人“SQL 跑起来了但落库一直报错”：你上游其实在产出更新流（比如聚合、join、去重），而你下游 JDBC 表却没主键。

2、版本与依赖：Flink 2.2 的现状先说清楚

在 Flink 2.2 的官方文档页面里，明确写了：Flink 2.2 还没有（yet）可用的 JDBC Connector 发布包。(nightlies.apache.org)

同时它也强调：JDBC connector 和各类 driver都不在 Flink 的二进制发行包里，集群跑任务时需要你自己把依赖带上（uber-jar 或放进 Flinklib/）。(nightlies.apache.org)

如果你当前不是强制 Flink 2.2，官网下载页能看到 JDBC Connector 的发布与兼容关系： (flink.apache.org)

• JDBC Connector 3.3.0：兼容 Flink 1.19.x / 1.20.x
• JDBC Connector 4.0.0：兼容 Flink 2.0.x

驱动方面，文档列了常见的 driver 坐标（MySQL / Oracle / PostgreSQL / Derby / SQL Server）。(nightlies.apache.org)

3、5 分钟上手：建表、读写、维表 Join 一次跑通

下面用文档里的 MySQL 例子做一条“最短路径”。

3.1 注册 JDBC 表（建议一定要声明主键）

CREATETABLEMyUserTable(idBIGINT,name STRING,ageINT,statusBOOLEAN,PRIMARYKEY(id)NOTENFORCED)WITH('connector'='jdbc','url'='jdbc:mysql://localhost:3306/mydatabase','table-name'='users');

3.2 写入 JDBC 表（把另一张表 T 的结果落库）

INSERTINTOMyUserTableSELECTid,name,age,statusFROMT;

3.3 扫描读取

SELECTid,name,age,statusFROMMyUserTable;

3.4 当维表做 temporal join（Lookup Source，同步查询）

SELECT*FROMmyTopicLEFTJOINMyUserTableFORSYSTEM_TIMEASOFmyTopic.proctimeONmyTopic.key=MyUserTable.id;

以上 DDL 与用法均来自官方文档示例。(nightlies.apache.org)

4、关键参数怎么选：按“真实生产场景”拆开讲

JDBC Connector 的参数很多，但真正影响大的主要集中在四块：连接、Scan、Lookup、Sink。(nightlies.apache.org)

4.1 连接与容错

• url/table-name：必填
• driver：可选，不填通常能从 url 推导
• username/password：可选，但要么都填要么都不填
• connection.max-retry-timeout：最大重试间隔上限（默认 60s）(nightlies.apache.org)

4.2 批量读取 Scan：fetch-size + auto-commit（Postgres 重点关注）

• scan.fetch-size：每次 round-trip 拉多少行（0 表示忽略 hint）(nightlies.apache.org)
• scan.auto-commit：默认 true；但文档特别点名Postgres 可能需要设为 false 才能流式读取结果（否则容易一次性把结果集全拉到内存侧）。(nightlies.apache.org)

4.3 大表加速：Partitioned Scan（并行扫描的正确姿势）

当你批读上亿行时，单并发 JDBC 扫描基本等于“用吸管喝海”。JDBC Connector 支持按区间切分，让多个 source 并行拉取：

必须同时配置这四个（配置任意一个就要配齐其他三个）：(nightlies.apache.org)

• scan.partition.column：必须是 numeric / date / timestamp 列
• scan.partition.num：分区数
• scan.partition.lower-bound：最小值
• scan.partition.upper-bound：最大值

实操建议：

• partition.column 优先选“分布均匀、单调增长”的列（自增 id、业务时间戳）
• lower/upper 在批任务里可以先查一遍 MIN/MAX 再提交作业（文档也提示这是可行路径）。(nightlies.apache.org)

4.4 维表 Join 性能：Lookup Cache（PARTIAL）

JDBC 维表 Join 最大的问题是：每条流都打一次 DB。Connector 提供了进程级缓存（每个 TaskManager 一份），核心开关是：

• lookup.cache = PARTIAL
• lookup.partial-cache.max-rows
• lookup.partial-cache.expire-after-write
• lookup.partial-cache.expire-after-access
• lookup.partial-cache.cache-missing-key（默认 true：连“查不到”的空结果也缓存）
• lookup.max-retries（默认 3）(nightlies.apache.org)

注意这就是典型的“吞吐 vs 新鲜度”权衡：TTL 越短越新鲜，但 DB 压力越大；TTL 越长吞吐越高，但维表可能偏旧。(nightlies.apache.org)

4.5 写入侧：buffer flush 决定延迟与吞吐

写入 JDBC 时，Connector 会做缓冲 + 异步 flush：

• sink.buffer-flush.max-rows：攒多少行刷一次（可设 0 禁用）
• sink.buffer-flush.interval：多久必须刷一次（可设 0 禁用）
• sink.max-retries：写失败重试次数
• sink.parallelism：sink 并行度（默认跟上游）(nightlies.apache.org)

直觉版调参：

• 追低延迟：interval 小一点、max-rows 小一点
• 追高吞吐：max-rows 大一点、sink 并行度拉起来（同时关注 DB 承载）

5、幂等与 Upsert：主键是你的“安全带”

文档明确说明：Flink 写外部 DB 时，会使用 DDL 里声明的主键。(nightlies.apache.org)

• Upsert 模式：按主键插入或更新，具备更强幂等性（失败恢复、数据重放都更稳）
• Append 模式：所有记录都当 INSERT，遇到唯一键/主键冲突就可能失败 (nightlies.apache.org)

同时因为各家数据库 upsert 语法不同，Connector 会用数据库方言生成对应 DML（官方给了映射表）：

• MySQL：INSERT .. ON DUPLICATE KEY UPDATE ..
• PostgreSQL：INSERT .. ON CONFLICT .. DO UPDATE SET ..
• Oracle：MERGE INTO ..
• SQL Server：MERGE INTO ..(nightlies.apache.org)

结论很工程化：

• 强烈建议 JDBC sink 表定义主键，并确保这个主键在目标库确实是主键或唯一键集合，否则你以为自己在 upsert，实际可能是在不停撞约束。

6、JdbcCatalog：把数据库“直接变成 Catalog”，少写一堆 DDL

JdbcCatalog 允许你通过 JDBC 把外部数据库挂到 Flink Catalog 下。目前文档说明主要提供 Postgres Catalog 与 MySQL Catalog 两种实现，支持的 catalog 方法也有限。(nightlies.apache.org)

创建 Catalog（文档示例结构）：

CREATECATALOG my_catalogWITH('type'='jdbc','default-database'='mydb','username'='...','password'='...','base-url'='jdbc:postgresql://<ip>:<port>');USECATALOG my_catalog;

PostgreSQL 的 schema 映射要特别注意：默认 schema 是public，如果访问自定义 schema，需要用反引号把schema.table整体转义。(nightlies.apache.org)
MySQL 则更直接：db 与表是平铺关系，默认库来自创建 catalog 时的default-database。(nightlies.apache.org)

7、类型映射：DDL 不知道怎么写时看这张表

文档给了 MySQL / Oracle / PostgreSQL / SQL Server 到 Flink SQL 的类型映射（int/bigint/decimal/boolean/date/time/timestamp/string/bytes/array 等）。遇到建表报类型不兼容时，优先对照这张表去改字段类型，而不是盲目改 cast。(nightlies.apache.org)

8、上线前自检清单（很实用）

• 你的 JDBC sink DDL 有PRIMARY KEY吗？上游会不会产生更新流？(nightlies.apache.org)
• 批读大表是否启用 Partitioned Scan？四个分区参数是否配齐？partition.column 是否合适？(nightlies.apache.org)
• Postgres 大结果集读取是否需要scan.auto-commit = false？(nightlies.apache.org)
• 维表 Join 是否开启 PARTIAL 缓存？TTL 是否满足业务“新鲜度”预期？(nightlies.apache.org)
• 写入侧 flush 节奏（max-rows/interval）是否匹配你的延迟/吞吐目标？重试次数是否合理？(nightlies.apache.org)
• connector jar 与 driver jar 是否随作业一起带到集群？（它们不在 Flink 二进制发行包里）(nightlies.apache.org)\