在开发Flink SQL时报错”Exception in thread “main” java.lang.IllegalStateException: please declare primary key for sink table when query contains update/delete record”

这个错误表明您的Flink作业涉及更新或删除操作（如CDC场景），而JDBC Sink表没有定义主键。以下是详细分析和解决方案：

错误原因分析

当Flink作业涉及UPDATE/DELETE操作时（常见于CDC场景或使用Retract流），JDBC Sink表必须显式声明主键，因为：

1. 主键用于定位目标数据库中需要更新的记录
2. 没有主键时，Flink无法确定如何执行更新/删除操作

解决方案

在表定义中添加 PRIMARY KEY 声明（需与目标数据库表主键一致），例如：

val sinkDDL =
  """
    |CREATE TABLE dws_iot_extreme_info_min (
    |  device_code STRING,
    |  `day` DATE,
    |  ptime BIGINT,
    |  station_sn STRING,
    |  unit STRING,
    |  cu_device_code STRING,
    |  max_u_value DECIMAL(5,3),
    |  min_u_value DECIMAL(5,3),
    |  max_t_value DECIMAL(5,2),
    |  min_t_value DECIMAL(5,2),
    |  max_soc_value DECIMAL(5,2),
    |  min_soc_value DECIMAL(5,2),
    |  max_u_time BIGINT,
    |  min_u_time BIGINT,
    |  max_t_time BIGINT,
    |  min_t_time BIGINT,
    |  max_soc_time BIGINT,
    |  min_soc_time BIGINT,
    |  dt DATE,
    |  PRIMARY KEY (device_code, `day`) NOT ENFORCED  -- 添加主键声明
    |) WITH (
    |  'connector' = 'jdbc',
    |  'url' = 'jdbc:mysql://10.0.2.2:3306/cnsaas',
    |  'table-name' = 'dws_bigdata_device_extreme_info_min',
    |  'driver' = 'com.mysql.cj.jdbc.Driver',
    |  'username' = 'root',
    |  'password' = '',
    |  'sink.buffer-flush.max-rows' = '1000',    
    |  'sink.buffer-flush.interval' = '1s',      
    |  'sink.max-retries' = '3'                 
    |)
    |""".stripMargin

关键修改点说明

1. 主键声明：PRIMARY KEY (device_code, `day`) NOT ENFORCED
   • 主键字段需与目标数据库表的主键一致
   • NOT ENFORCED 表示Flink不会校验数据主键约束，由数据库负责
2. 目标表要求：
   • MySQL数据库中 dws_iot_extreme_info_min 表必须有相同的主键定义
   • 可通过以下SQL确保主键存在：
   • ALTER TABLE dws_iot_extreme_info_min ADD PRIMARY KEY (device_code, day);