大佬们现在咋处理flink cdc interval join场景的，现在测试下来一直报？

在使用 Flink CDC 进行 Interval Join 时，如果遇到报错 "StreamPhysicalIntervalJoin doesn't support consuming update and delete changes which is produced by node TableSourceScan"，这通常是因为 Interval Join 算子无法处理上游产生的更新（update）和删除（delete）变更事件。以下是详细的分析和解决方案：

问题原因

1. Interval Join 的限制
   Interval Join 是一种基于时间范围的 Join 操作，当前 Flink 的实现不支持消费上游表中产生的更新（update）和删除（delete）变更事件。

   • 如果上游表是通过 CDC（如 MySQL Binlog 或 Debezium）读取的，CDC 数据流会包含插入（insert）、更新（update）和删除（delete）三种类型的变更事件。
   • 当这些变更事件传递到 Interval Join 时，由于其内部实现的限制，无法正确处理 update 和 delete 类型的变更。

2. 数据流类型不匹配
   如果上游表的数据流是 Append-only 类型（即只包含插入操作），则不会触发该问题。但如果数据流是 Changelog 流（包含 insert、update、delete），就会导致报错。

解决方案

方案一：将上游表转换为 Append-only 数据流

如果业务场景允许忽略更新和删除操作，可以通过以下方式将上游表的数据流转换为 Append-only 类型： 1. 过滤掉 update 和 delete 事件
在定义上游表时，可以使用 ROW_KIND 过滤掉非插入事件。例如：

CREATE TABLE source_table (
    id BIGINT,
    name STRING,
    ts TIMESTAMP(3),
    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'your-hostname',
    'port' = '3306',
    'username' = 'your-username',
    'password' = 'your-password',
    'database-name' = 'your-database',
    'table-name' = 'your-table'
);

-- 过滤掉 update 和 delete 事件
CREATE VIEW filtered_source AS
SELECT * FROM source_table WHERE ROW_KIND = 'INSERT';

然后在 Interval Join 中使用 filtered_source 表。

1. 使用 Upsert Kafka 或其他支持 Changelog 的中间存储
   将上游表的数据先写入一个支持 Changelog 的中间存储（如 Upsert Kafka），然后从中间存储中读取 Append-only 数据流进行 Join 操作。

方案二：避免使用 Interval Join

如果业务场景允许，可以考虑使用其他类型的 Join 操作替代 Interval Join： 1. Temporal Join
Temporal Join 支持基于时间版本的 Join 操作，适用于需要关联历史数据的场景。例如：

SELECT o.id, o.product_name, s.status
FROM Orders AS o
JOIN Shipments FOR SYSTEM_TIME AS OF o.proc_time AS s
ON o.id = s.orderId;

注意：Temporal Join 同样存在状态迁移的限制，需确保 Schema 和条件兼容。

1. 普通 Join
   如果时间范围的约束不是必须的，可以使用普通的 Inner Join 或 Left Join 替代。

方案三：升级 Flink 版本

Flink 社区和阿里云实时计算团队持续优化 CDC 和 Join 的功能。建议检查当前使用的 Flink 版本是否为最新版本，并升级到支持更复杂场景的版本（如 VVR 4.0.13 及以上）。新版本可能已经解决了部分限制问题。

方案四：调整作业逻辑

如果上述方法均不可行，可以尝试调整作业逻辑，避免直接在 Interval Join 中处理 CDC 数据流： 1. 预处理 CDC 数据
在进入 Interval Join 前，对 CDC 数据进行预处理，生成一个只包含所需字段和事件类型的中间表。 2. 拆分 Join 操作
将复杂的 Join 操作拆分为多个简单的步骤，逐步处理数据流。

重要提醒

• Binlog 格式要求
  确保 MySQL 的 Binlog 格式设置为 ROW，否则可能导致 CDC 数据流无法正确解析。

  SET GLOBAL binlog_format=ROW;
  

• 权限配置
  确保用于读取 CDC 数据的数据库用户具有足够的权限，包括 SELECT、REPLICATION SLAVE 和 REPLICATION CLIENT。

• 资源分配
  如果作业处理速度较慢，可能导致 Binlog 被清理，建议增加 Binlog 的保留时间或分配更多资源给 Flink 作业。

通过以上方法，您可以有效解决 StreamPhysicalIntervalJoin 不支持消费 update 和 delete 变更事件 的问题。根据实际业务需求选择合适的方案，并确保环境配置符合要求。