Flink cdc 2.4 在使用Flink sql的方式 读mysql，如何过滤回撤流？

在使用 Flink CDC 2.4 和 Flink SQL 读取 MySQL 数据时，如果结果涉及聚合函数类型的 SQL，并希望过滤回撤流（retract stream），只输出最终的结果，可以通过以下方法实现：

1. 理解回撤流的机制

Flink 在处理聚合操作时，默认会生成回撤流（retract stream）。回撤流包含两种类型的消息： - INSERT/UPDATE_AFTER：表示新增或更新后的数据。 - DELETE/UPDATE_BEFORE：表示删除或更新前的数据。

为了只输出最终的结果，需要过滤掉 DELETE 和 UPDATE_BEFORE 类型的消息，仅保留 INSERT 和 UPDATE_AFTER 类型的消息。

2. 使用 changelog-producer 参数

Paimon 表支持通过 changelog-producer 参数控制增量数据的产生机制。如果需要过滤回撤流并只输出最终结果，可以将 changelog-producer 设置为 lookup 或 full-compaction。具体配置如下：

CREATE TABLE MyTable (
    product_id BIGINT,
    price DOUBLE,
    sales BIGINT,
    PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
    'merge-engine' = 'aggregation',
    'fields.price.aggregate-function' = 'max',
    'fields.sales.aggregate-function' = 'sum',
    'changelog-producer' = 'lookup' -- 或 'full-compaction'
);

• lookup 模式：通过查找表中的最新状态来生成最终结果。
• full-compaction 模式：通过对全量数据进行压缩来生成最终结果。

这两种模式都可以确保输出的流中只包含最终的聚合结果，而不会包含回撤消息。

3. 配置字段忽略回撤消息

如果某些字段不需要处理回撤消息，可以通过设置 fields.${field_name}.ignore-retract 参数来忽略回撤消息。例如：

WITH (
    'fields.price.ignore-retract' = 'true',
    'fields.sales.ignore-retract' = 'true'
)

此配置会确保指定字段在处理过程中忽略回撤消息，从而只输出最终的聚合结果。

4. 使用 INSERT INTO 输出到结果表

在 Flink SQL 中，可以通过 INSERT INTO 将聚合结果写入目标表。如果目标表是 MySQL 表，可以利用 ON DUPLICATE KEY UPDATE 语句来合并数据。示例如下：

INSERT INTO mysql_result_table
SELECT 
    product_id,
    MAX(price) AS max_price,
    SUM(sales) AS total_sales
FROM mysql_source_table
GROUP BY product_id;

在此过程中，MySQL 表会自动处理主键冲突，并更新对应的数据。

5. 注意事项

• 聚合函数的支持：并非所有聚合函数都支持回撤与删除操作。例如，SUM 支持回撤，但 MAX、MIN 等函数不支持。因此，在设计 SQL 时需注意选择合适的聚合函数。
• 性能优化：如果数据量较大，建议增加并发度或调整 Checkpoint 间隔时间，以避免全量读取阶段出现 OOM 问题。
• 时间字段的单调性：如果排序字段的更新不是单调的，可能会导致 TopN 优化算法无法生效。因此，需确保排序字段的更新方向与排序方向一致。

总结

通过上述方法，您可以有效过滤回撤流并只输出最终的聚合结果。关键步骤包括： 1. 配置 changelog-producer 参数为 lookup 或 full-compaction。 2. 忽略不必要的回撤消息（通过 ignore-retract 参数）。 3. 使用 INSERT INTO 将结果写入目标表，并利用 MySQL 的 ON DUPLICATE KEY UPDATE 机制合并数据。

这些方法能够帮助您在 Flink CDC 场景下高效处理聚合结果，同时避免回撤流对最终输出的影响。