说明：Flink CDC 底层基于 Debezium，下面是自动映射规则；Flink DECIMAL 最大精度 38，MySQL DECIMAL 最高 65，超过 38 必须手动映射 STRING 防丢精度Apache InL...

1. 整数类型

MySQL 类型	Flink SQL 类型	备注
TINYINT	TINYINT
TINYINT UNSIGNED / TINYINT UNSIGNED ZEROFILL	SMALLINT
SMALLINT	SMALLINT
SMALLINT UNSIGNED / SMALLINT UNSIGNED ZEROFILL	INT
INT / MEDIUMINT	INT
INT UNSIGNED / INT UNSIGNED ZEROFILL / MEDIUMINT UNSIGNED	BIGINT
BIGINT	BIGINT
BIGINT UNSIGNED / BIGINT UNSIGNED ZEROFILL / SERIAL	DECIMAL(20,0)	⚠️高频踩坑，无符号大数不是 BIGINT
TINYINT(1) / BIT(1) / BOOLEAN	BOOLEAN	可通过参数关闭 `treat-tinyint1-as-boolean`

2. 浮点 & 高精度小数

MySQL 类型	Flink SQL 类型	备注
FLOAT	FLOAT
DOUBLE / DOUBLE PRECISION / REAL	DOUBLE
DECIMAL(p,s) / NUMERIC(p,s)	DECIMAL(p,s)	p ≤ 38
DECIMAL(p,s)	STRING	p > 38，必须手动改

3. 时间日期（最容易时区踩坑）

MySQL 类型	Flink SQL 类型	备注
DATE	DATE
TIME[(p)]	TIME[(p)]
DATETIME[(p)]	TIMESTAMP[(p)]	无时区，字面时间
TIMESTAMP[(p)]	TIMESTAMP_LTZ[(p)]	带本地时区，MySQL 存 UTC，会受`server-time-zone`影响
YEAR	INT

✅ 重点：MySQL TIMESTAMP 映射为 TIMESTAMP_LTZ；DATETIME 映射普通 TIMESTAMP（不带时区），同步时时区参数必须对齐阿里云帮助...

4. 字符串、二进制、文本

MySQL 类型	Flink SQL 类型
CHAR(n)	CHAR(n)
VARCHAR(n)	VARCHAR(n)
TINYTEXT / TEXT / MEDIUMTEXT / LONGTEXT	STRING
BINARY(n)	BINARY(n)
VARBINARY(n)	VARBINARY(n)
TINYBLOB / BLOB / MEDIUMBLOB / LONGBLOB	BYTES
BIT(n)（n>1）	BINARY(ceil((n+7)/8))

5. 枚举、集合、JSON、空间类型

MySQL 类型	Flink SQL 类型	备注
ENUM	STRING	存枚举字面量
SET	ARRAY<STRING>	自动拆成字符串数组
JSON	STRING	JSON 原样序列化字符串，Flink1.16 + 可转 JSON 类型（手动 CAST）
GEOMETRY / POINT / LINESTRING 等空间类型	STRING	WKT/GeoJSON 字符串

高频踩坑清单（生产必看）

BIGINT UNSIGNED → DECIMAL(20,0)，如果下游是 Doris/ClickHouse，DDL 不要写 BIGINT，会溢出报错
TINYINT (1) 默认转 BOOLEAN，业务存 0/1 标识、不想转布尔：

source:
  connector: mysql-cdc
  treat-tinyint1-as-boolean: false

TIMESTAMP_LTZ 时区问题：source 配置 server-time-zone: Asia/Shanghai，否则时间偏移 8 小时
DECIMAL 精度 > 38，自动映射会报错，需要在 Flink DDL 显式定义为 STRING
MySQL JSON 只是 STRING，查询时需要 JSON_VALUE(json_col, '$.xx') 解析，Flink SQL 有 JSON 函数支持
SET 类型映射 ARRAY，写入很多下游库（如 MySQL、Doris）不支持数组，需要额外拆分

示例 Flink CDC Source DDL（对照映射）

CREATE TABLE mysql_source (
  id BIGINT,
  u_bigint DECIMAL(20,0), -- bigint unsigned
  status BOOLEAN, -- tinyint(1)
  dt TIMESTAMP(3), -- datetime(3)
  ts TIMESTAMP_LTZ(3), -- mysql timestamp(3)
  price DECIMAL(18,2),
  tags ARRAY<STRING>, -- set('a','b')
  ext STRING, -- json
  PRIMARY KEY(id) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = '127.0.0.1',
  'port' = '3306',
  'username' = 'root',
  'password' = 'xxx',
  'database-name' = 'test_db',
  'table-name' = 't_test',
  'server-time-zone' = 'Asia/Shanghai'
);