Flink CDC有没有办法配置。不接受dml 呀？或者遇到dml 语句忽略错误呢？

根据您提供的问题和知识库中的相关资料，以下是针对问题的详细解答：

1. Flink CDC是否可以配置为不接受DML，或者遇到DML语句时忽略错误？

在Flink CDC中，默认情况下会捕获数据库的变更数据（包括INSERT、UPDATE、DELETE等DML操作）。如果需要配置Flink CDC以忽略某些DML语句或避免处理特定类型的变更事件，可以通过以下方式实现：

方案一：通过schema.change.behavior配置忽略Schema变更

在Flink CDC的pipeline模块中，可以通过schema.change.behavior参数来控制Schema变更的行为。如果希望忽略某些DML语句导致的错误，可以将该参数设置为IGNORE模式。具体配置如下：

pipeline:
  schema.change.behavior: IGNORE

• LENIENT（默认）：尝试将Schema变更转换为目标端可处理的形式。
• EXCEPTION：不允许任何Schema变更行为，遇到变更时抛出异常。
• IGNORE：忽略所有Schema变更行为，不会对目标端产生影响。

通过设置为IGNORE，可以避免因DML语句引发的错误。

方案二：过滤特定的变更事件

如果需要更细粒度地控制哪些DML语句被处理，可以在Flink SQL中使用WHERE条件过滤掉不需要的变更事件。例如，仅处理INSERT事件，而忽略UPDATE和DELETE事件：

CREATE TABLE source_table (
  id BIGINT,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = '<your-host>',
  'port' = '3306',
  'username' = '<your-username>',
  'password' = '<your-password>',
  'database-name' = '<your-database>',
  'table-name' = '<your-table>'
);

CREATE TABLE sink_table (
  id BIGINT,
  name STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'topic' = 'output_topic',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'json'
);

-- 仅处理INSERT事件
INSERT INTO sink_table
SELECT id, name
FROM source_table
WHERE ROW_KIND = 'INSERT';

通过ROW_KIND字段过滤，可以实现对特定DML语句的忽略。

2. 现在配置成ROW模式系统表还是有MIX类型过来，帮忙看下吧

根据知识库中的信息，MySQL的Binlog格式需要明确设置为ROW模式，否则可能会出现Mixed格式的事件，导致Flink CDC无法正确解析。以下是排查和解决此问题的步骤：

步骤一：确认MySQL Binlog格式

执行以下SQL命令，检查当前MySQL服务器的Binlog格式：

SHOW VARIABLES LIKE 'binlog_format';

如果返回值为MIXED或STATEMENT，需要将其修改为ROW模式：

SET GLOBAL binlog_format=ROW;

注意：上述命令仅对当前会话生效。如果需要永久生效，请在MySQL配置文件（如my.cnf）中添加以下内容：

[mysqld]
binlog_format=ROW

然后重启MySQL服务。

步骤二：验证从库配置

如果您的Flink CDC连接的是MySQL从库，确保从库已正确配置以接收主库同步的数据，并将这些数据写入从库自身的Binlog文件中。需要启用以下参数：

log-slave-updates = 1
gtid_mode = ON
enforce_gtid_consistency = ON

步骤三：清理历史Binlog

如果MySQL服务器上存在Mixed格式的历史Binlog，可能会导致Flink CDC作业继续接收到Mixed格式的事件。建议清理历史Binlog并重新启动CDC作业：

1. 检查Binlog保留时间：

   SHOW VARIABLES LIKE 'expire_logs_days';
   

2. 增加Binlog保留时间（例如设置为7天）：

   SET GLOBAL expire_logs_days=7;
   

3. 手动清理旧的Binlog文件：

   PURGE BINARY LOGS BEFORE 'YYYY-MM-DD HH:MM:SS';
   

步骤四：调整Debezium配置

如果仍然遇到Mixed格式的事件，可以通过Debezium的配置参数进一步处理。例如，设置debezium.inconsistent.schema.handling.mode为warn，以避免因Mixed格式事件导致的报错：

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

总结

1. 如果需要忽略DML语句或避免处理特定变更事件，可以通过schema.change.behavior设置为IGNORE，或在Flink SQL中使用ROW_KIND过滤。
2. 针对Mixed格式的问题，需确保MySQL的Binlog格式设置为ROW模式，并验证从库配置是否正确。同时，清理历史Binlog和调整Debezium配置也能有效解决问题。

希望以上解答能够帮助您解决问题！如有其他疑问，请随时联系。