Flink CDC创建表之后，再查询就报这个错是为什么？

在使用Flink CDC创建表后，查询时出现错误可能与以下几个原因相关。以下是详细的分析和解决方案：

1. 表结构变更导致的同步问题

如果在Flink CDC全量阶段发生了表结构变更（例如新增列、修改列类型等），可能会导致作业报错或无法同步表结构变更。

原因分析

• Flink CDC在全量阶段会读取表的快照数据，而在增量阶段会消费Binlog。如果在全量阶段表结构发生变更，可能导致快照数据与Binlog数据不一致。
• 如果表结构变更是不支持的类型（如删除主键、修改主键列等），会导致同步失败。

解决方案

1. 停止当前Flink作业。
2. 删除下游目标表（确保数据可以重新同步）。
3. 无状态地重新启动Flink作业（即清除作业的状态信息）。
4. 避免在同步过程中进行不兼容的表结构变更。

2. 权限不足导致的查询失败

如果使用的数据库用户没有足够的权限访问表或执行某些操作，也可能导致查询失败。

常见报错示例

• DELETE command denied to user 'userName'@'*.*.*.*' for table 'table_name'
• sub account not auth permission

解决方案

1. 检查SQL逻辑是否存在需要DELETE权限的操作（如CDC流结合WHERE条件过滤时，UPDATE_BEFORE会被识别为DELETE操作）。
2. 确保数据库用户具有以下权限：
   • SELECT：用于读取表数据。
   • REPLICATION SLAVE 和 REPLICATION CLIENT：用于读取Binlog。
   • DELETE：如果存在UPDATE_BEFORE或DELETE操作。
3. 如果使用的是RDS MySQL，确保RAM用户具有下载OSS上Binlog文件的权限。

3. Binlog相关问题

如果MySQL服务器上的Binlog被清理或配置不当，也可能导致查询失败。

常见报错示例

• The connector is trying to read binlog starting at GTIDs ..., but this is no longer available on the server
• Encountered change event for table xxx.xxx whose schema isn't known to this connector

原因分析

• MySQL服务器上的Binlog保留时间过短（默认为1天），导致Flink CDC无法读取所需的Binlog。
• Binlog过滤器配置（如Binlog_Do_DB或Binlog_Ignore_DB）忽略了某些库的Binlog。

解决方案

1. 检查并调整Binlog保留时间：

   SHOW VARIABLES LIKE 'expire_logs_days';
   SET GLOBAL expire_logs_days=7;
   

2. 检查Binlog过滤器配置：

   SHOW MASTER STATUS;
   

   确保目标库未被Binlog_Ignore_DB忽略，且Binlog_Do_DB包含目标库。

4. Schema不匹配或解析失败

如果Flink CDC无法解析表的Schema，也可能导致查询失败。

常见报错示例

• Encountered change event for table xxx.xxx whose schema isn't known to this connector
• The "before" field of UPDATE/DELETE message is null

原因分析

• 数据库用户权限不足，导致无法获取完整的Schema信息。
• 使用了debezium.snapshot.mode='never'，导致从Binlog开头读取数据，但Binlog中的Schema与当前表的Schema不匹配。
• Debezium无法解析某些特定的DDL语句（如DEFAULT (now())）。

解决方案

1. 确保数据库用户具有所有相关表的权限。
2. 避免使用debezium.snapshot.mode='never'，可以通过以下参数避免报错：

   debezium.inconsistent.schema.handling.mode = 'warn'
   

3. 检查日志中io.debezium.connector.mysql.MySqlSchema WARN的详细信息，定位无法解析的变更，并根据提示调整表结构。

5. 主键缺失

如果表未定义主键，且启用了增量快照功能（scan.incremental.snapshot.enabled），也会导致查询失败。

常见报错示例

• The primary key is necessary when enable 'Key: 'scan.incremental.snapshot.enabled'

解决方案

在DDL中为表添加主键（Primary Key）信息。例如：

CREATE TABLE your_table (
    id INT PRIMARY KEY,
    name STRING
) WITH (
    'connector' = 'mysql-cdc',
    ...
);

6. 其他可能的原因

• 重复事件投递：在非正常情况下（如故障恢复），Maxwell可能投递重复的变更事件，导致Flink查询结果异常。
  解决方案：设置table.exec.source.cdc-events-duplicate=true，并在源表上定义主键以去重。
• 网络连通性问题：确保MySQL与Flink在相同VPC下，跨VPC或公网访问时需打通网络。

总结

根据上述分析，建议按照以下步骤排查问题： 1. 检查是否有表结构变更，并按需重新同步表数据。 2. 确保数据库用户具有足够的权限。 3. 调整Binlog保留时间和过滤器配置。 4. 检查Schema是否匹配，必要时重新生成快照。 5. 确保表定义中包含主键信息。

如果问题仍未解决，请提供具体的报错信息以便进一步分析。