Flink CDC 中canal-json和debizum-json，大家都是用哪个？

在Flink CDC中，canal-json和debezium-json是两种常见的数据格式，用于捕获数据库的变更日志（Change Data Capture, CDC）。选择使用哪种格式通常取决于具体的业务需求、技术栈以及对工具的熟悉程度。以下是两种格式的特点及适用场景分析：

1. Canal-JSON

特点：

• 来源：Canal 是阿里巴巴开源的CDC工具，主要用于捕获MySQL的变更日志。
• 默认序列化方式：支持JSON和Protobuf，默认使用Protobuf，但JSON格式更常见。
• 消息结构：
  • 包含data字段（更新后的数据）和old字段（更新前的数据）。
  • 提供元信息如database、table、type（操作类型，如INSERT、UPDATE、DELETE）等。
  • 示例：

  {
    "data": [{"id": "1", "name": "Alice"}],
    "old": [{"name": "Bob"}],
    "type": "UPDATE",
    "database": "test_db",
    "table": "users"
  }
  

• 配置选项：
  • 支持处理Map中空key值的模式（map-null-key.mode），可选值为FAIL、DROP或LITERAL。
  • 支持时间戳格式（timestamp-format.standard），可选值为SQL或ISO-8601。
  • 支持正则匹配指定数据库或表的变更记录（database.include、table.include）。

适用场景：

• MySQL生态：Canal专为MySQL设计，适合需要从MySQL同步数据到其他系统的场景。
• 简单易用：对于熟悉阿里巴巴技术栈的用户，Canal的JSON格式更容易上手。
• 实时性要求高：Canal能够快速捕获MySQL的Binlog变更，适合实时数据同步和增量数据处理。

2. Debezium-JSON

特点：

• 来源：Debezium 是一个开源的分布式CDC平台，支持多种数据库（如MySQL、PostgreSQL、MongoDB等）。
• 消息结构：
  • 包含before字段（更新前的数据）和after字段（更新后的数据）。
  • 提供丰富的元信息，如source字段（包含数据库名、表名、时间戳等）。
  • 示例：

  {
    "before": {"id": 1, "name": "Bob"},
    "after": {"id": 1, "name": "Alice"},
    "op": "u",
    "source": {
      "db": "test_db",
      "table": "users",
      "ts_ms": 1728528674000
    }
  }
  

• 配置选项：
  • 支持是否包含schema信息（schema-include），可选值为true或false。
  • 支持处理Map中空key值的模式（map-null-key.mode），与Canal类似。
  • 支持DECIMAL类型的科学计数法表示（encode.decimal-as-plain-number）。

适用场景：

• 多数据库支持：Debezium支持多种数据库，适合需要跨数据库进行数据同步的场景。
• 复杂数据处理：Debezium的消息结构更加丰富，适合需要详细元信息的场景（如审计日志、实时物化视图）。
• 社区活跃：Debezium拥有广泛的社区支持，适合需要长期维护和扩展的项目。

3. 使用情况对比

根据知识库中的描述，以下是一些实际使用情况的对比： - 默认格式：在阿里云的Flink CDC场景中，debezium-json通常是默认格式，尤其是在Kafka数据源中。 - 兼容性：debezium-json支持更多的数据库类型，而canal-json主要针对MySQL。 - 性能：两者在性能上差异不大，但在特定场景下，canal-json可能更适合MySQL生态。

4. 推荐选择

• 如果您的系统主要基于MySQL，并且需要快速实现数据同步，建议使用Canal-JSON。
• 如果您的系统需要支持多种数据库，或者需要更丰富的元信息，建议使用Debezium-JSON。

5. 配置示例

Canal-JSON 配置示例：

CREATE TABLE KafkaTable (
  id BIGINT,
  name STRING,
  description STRING,
  weight DECIMAL(10, 2)
) WITH (
  'connector' = 'kafka',
  'topic' = 'products_binlog',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'canal-json'
);

Debezium-JSON 配置示例：

CREATE TABLE KafkaTable (
  id BIGINT,
  name STRING,
  description STRING,
  weight DECIMAL(10, 2)
) WITH (
  'connector' = 'kafka',
  'topic' = 'products_binlog',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format' = 'debezium-json'
);

重要提醒

• 分库分表场景：如果单表数据分布在多个分区中，需将distributed-tables配置项设为true。
• Schema推导策略：Kafka数据源支持多种Schema推导策略，可通过schema.inference.strategy配置。

通过以上分析，您可以根据具体需求选择合适的格式。