flink cdc 读取mysql数据写入hudi表中，为什么出现这种情况？

问题分析

根据您的描述，Flink CDC 从 MySQL 源表读取数据并写入第一个 Hudi 表（Hudi Table A），然后通过 Flink SQL 查询 Hudi Table A 的数据并写入第二个 Hudi 表（Hudi Table B）。当源表中的数据被删除时，Hudi Table A 中的数据成功删除，但 Hudi Table B 中的数据未同步删除。

此问题的核心在于 Hudi 表的 UPSERT 和 DELETE 语义 以及 Flink SQL 在处理 Hudi 表时的行为。以下是详细分析和解决方案。

可能原因及解决方案

1. Hudi 表的 DELETE 语义传播问题

Hudi 支持 ACID 语义，默认提供 SNAPSHOT ISOLATION 隔离级别，并支持 UPSERT 和 DELETE 操作。然而，DELETE 操作的传播依赖于下游任务是否正确处理了删除标记（Delete Marker）。

• 原因：在 Flink SQL 查询 Hudi Table A 并写入 Hudi Table B 的过程中，可能未正确传递删除标记。Hudi 的删除操作是通过在数据中插入一条带有 is_deleted 标记的记录实现的。如果下游任务未识别或未处理该标记，则 Hudi Table B 中的数据不会被删除。

• 解决方案：

  • 确保 Flink SQL 查询 Hudi Table A 时启用了 Changelog 模式（Changelog Mode），以便捕获包括删除操作在内的所有变更数据流。
  • 在写入 Hudi Table B 时，确保目标表的写入模式支持 DELETE 操作。例如，使用 upsert 写操作类型，并确保目标表的主键与源表一致。

2. Flink SQL 查询逻辑未处理删除事件

Flink SQL 默认会生成 Changelog 流，包含 INSERT、UPDATE 和 DELETE 事件。但如果查询逻辑未显式处理 DELETE 事件，则可能导致下游表无法感知删除操作。

• 原因：Flink SQL 查询 Hudi Table A 时，可能仅处理了 INSERT 和 UPDATE 事件，而忽略了 DELETE 事件。

• 解决方案：

  • 在 Flink SQL 查询中，显式处理 DELETE 事件。例如，使用 INSERT INTO 或 UPSERT INTO 语法时，确保目标表能够接收并处理删除标记。
  • 如果目标表不支持 DELETE 操作，可以通过自定义逻辑（如过滤条件）手动处理删除事件。

3. Hudi 表的索引机制导致数据未删除

Hudi 使用索引机制来保证数据的唯一性。如果 Hudi Table B 的索引配置不当，可能导致删除操作未能正确生效。

• 原因：Hudi Table B 的索引配置可能未正确映射到 Hudi Table A 的主键，导致删除操作无法匹配到对应记录。

• 解决方案：

  • 检查 Hudi Table B 的索引配置，确保其主键与 Hudi Table A 的主键一致。
  • 如果使用了自定义分区字段，确保分区字段的值在两个表中保持一致。

4. Flink 作业的 Checkpoint 和 State 配置问题

Flink 作业的状态管理和 Checkpoint 配置可能影响 DELETE 操作的传播。

• 原因：如果 Flink 作业的 Checkpoint 配置不当，可能导致 DELETE 事件在状态恢复时丢失。

• 解决方案：

  • 确保 Flink 作业启用了 Checkpoint，并设置了合理的间隔时间（如 1 分钟）。
  • 检查 Flink 作业的日志，确认是否有 DELETE 事件被正确捕获和处理。

具体操作步骤

以下是针对上述问题的具体操作步骤：

1. 启用 Changelog 模式

   • 在 Flink SQL 查询 Hudi Table A 时，确保启用了 Changelog 模式。例如：

     CREATE TABLE hudi_table_a (
      id BIGINT PRIMARY KEY NOT ENFORCED,
      name STRING,
      age INT
     ) WITH (
      'connector' = 'hudi',
      'path' = 'oss://your-bucket/hudi-table-a',
      'table.type' = 'COPY_ON_WRITE',
      'changelog.enabled' = 'true'
     );
     

2. 确保目标表支持 DELETE 操作

   • 在写入 Hudi Table B 时，使用 upsert 写操作类型，并确保目标表的主键与源表一致。例如：

     CREATE TABLE hudi_table_b (
      id BIGINT PRIMARY KEY NOT ENFORCED,
      name STRING,
      age INT
     ) WITH (
      'connector' = 'hudi',
      'path' = 'oss://your-bucket/hudi-table-b',
      'table.type' = 'COPY_ON_WRITE',
      'write.operation' = 'upsert'
     );
     

3. 检查索引配置

   • 确保 Hudi Table B 的索引配置与 Hudi Table A 一致。例如，使用相同的主键和分区字段。

4. 优化 Flink 作业配置

   • 启用 Checkpoint 并设置合理的间隔时间：

     execution.checkpointing.interval: 60000
     

   • 检查 Flink 作业日志，确认 DELETE 事件是否被正确捕获。

5. 验证数据一致性

   • 在 Hudi Table A 和 Hudi Table B 中分别查询数据，确认 DELETE 操作是否生效。例如：

     SELECT * FROM hudi_table_a WHERE id = 1;
     SELECT * FROM hudi_table_b WHERE id = 1;
     

总结

通过以上分析和操作步骤，您可以解决 Hudi 表中 DELETE 操作未同步的问题。关键点在于确保 Flink SQL 查询逻辑正确处理 DELETE 事件，并在目标表中启用支持 DELETE 操作的配置。如果问题仍未解决，建议检查 Flink 作业的日志和 Hudi 表的元数据，进一步定位问题根源。