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 = PARTIALlookup.partial-cache.max-rowslookup.partial-cache.expire-after-writelookup.partial-cache.expire-after-accesslookup.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)\
