Flink CDC中这里除了支持有支持canal json吗!大佬们 都用的什么版本啊？

在Flink CDC中，除了支持JsonDebeziumDeserializationSchema外，也支持Canal JSON格式的反序列化。以下是关于Canal JSON的支持情况及相关版本信息的详细说明：

1. Canal JSON 支持

Flink CDC 提供了对 Canal JSON 格式的支持，主要用于解析 MySQL 的变更日志（ChangeLog）。通过配置 canal-json 格式，可以将 Canal 的 JSON 消息解析为 Flink SQL 系统中的 INSERT、UPDATE 或 DELETE 消息。

配置示例

以下是一个使用 Canal JSON 格式的典型配置：

CREATE TABLE KafkaTable (
    origin_database STRING METADATA FROM 'value.database' VIRTUAL,
    origin_table STRING METADATA FROM 'value.table' VIRTUAL,
    user_id BIGINT,
    item_id BIGINT,
    behavior STRING
) WITH (
    'connector' = 'kafka',
    'topic' = 'user_behavior',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'testGroup',
    'scan.startup.mode' = 'earliest-offset',
    'value.format' = 'canal-json'
);

在此配置中，value.format 设置为 canal-json，表示使用 Canal JSON 格式解析数据。

额外配置项

• canal-json.ignore-parse-errors：当解析异常时，可以选择跳过当前字段或行（设置为 true），或者报错并导致作业启动失败（默认值为 false）。
• canal-json.timestamp-format.standard：指定时间戳格式，支持 SQL 和 ISO-8601 两种格式。
• canal-json.database.include 和 canal-json.table.include：通过正则表达式过滤特定数据库或表的变更记录。

2. 版本要求

根据知识库资料，以下版本信息与 Canal JSON 的支持相关： - VVR 8.0.11 及以上版本：支持 canal-json.distributed-tables 配置项，用于处理单张表数据分布在多个分区的场景。 - 实时计算引擎 VVR 3.0.0 及以上版本：支持 JSON 格式的增强功能，包括 Canal JSON 的解析能力。

3. 社区使用情况

在实际应用中，用户通常会根据具体需求选择合适的反序列化器： - 如果需要解析 Debezium 格式的变更日志，可以选择 JsonDebeziumDeserializationSchema。 - 如果使用 Canal 工具捕获 MySQL 的变更日志，则推荐使用 canal-json 格式。

4. 重要提醒

• 合并 UPDATE_BEFORE 和 UPDATE_AFTER：目前 Flink 不支持将 UPDATE_BEFORE 和 UPDATE_AFTER 合并为一条 UPDATE 消息，请注意此限制。
• 数值类型处理：如果使用 JsonDebeziumDeserializationSchema 时遇到数值类型显示为字符串的问题，可以通过配置 bigint.unsigned.handling.mode 和 decimal.handling.mode 参数解决。

总结

Flink CDC 支持 Canal JSON 格式，并且推荐使用 VVR 8.0.11 及以上版本以获得完整的功能支持。如果您正在使用 Canal 工具捕获 MySQL 的变更日志，建议直接配置 canal-json 格式进行数据解析。