Flink CDC 监听二进制主键 MySQL 表的解决方案

当使用 Flink CDC 监听主键为二进制格式(BINARY/VARBINARY/BLOB等)的 MySQL 表时，可能会遇到报错问题。以下是常见问题和解决方案：

常见错误

1. 主键类型不支持错误：Unsupported primary key type: BINARY
2. 反序列化错误：Failed to deserialize binary key
3. Schema 不匹配错误：Schema mismatch for binary key

解决方案

方案1：修改表结构（推荐）

如果可能，修改表结构将二进制主键转换为字符串类型：

ALTER TABLE your_table MODIFY COLUMN id VARCHAR(255) 
CHARACTER SET utf8mb4 COLLATE utf8mb4_bin;

方案2：使用自定义转换器

在 Flink CDC 配置中添加自定义反序列化器：

DebeziumDeserializationSchema<RowData> deserializer = new JsonDebeziumDeserializationSchema(true) {
    @Override
    public void deserialize(SourceRecord record, Collector<RowData> out) throws Exception {
        // 自定义处理二进制主键
        Struct value = (Struct) record.value();
        Struct after = value.getStruct("after");

        // 处理二进制字段
        byte[] binaryKey = after.getBytes("id");
        String stringKey = bytesToHex(binaryKey); // 转换为十六进制字符串

        // 创建新的RowData
        GenericRowData rowData = new GenericRowData(1);
        rowData.setField(0, stringKey);

        out.collect(rowData);
    }
};

MySQLSource.<RowData>builder()
    .hostname("localhost")
    .port(3306)
    .databaseList("your_db")
    .tableList("your_db.your_table")
    .username("user")
    .password("password")
    .deserializer(deserializer)
    .build();

方案3：使用 Flink SQL 转换

在 Flink SQL 中创建视图进行转换：

-- 原始表定义
CREATE TABLE source_table (
    id BINARY,
    other_columns STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'user',
    'password' = 'password',
    'database-name' = 'your_db',
    'table-name' = 'your_table'
);

-- 转换视图
CREATE VIEW converted_view AS
SELECT 
    CAST(id AS STRING) AS id_str,
    other_columns
FROM source_table;

方案4：配置 Debezium 属性

在 CDC 连接器配置中添加 Debezium 属性：

MySQLSource.<RowData>builder()
    .hostname("localhost")
    .port(3306)
    .databaseList("your_db")
    .tableList("your_db.your_table")
    .username("user")
    .password("password")
    .deserializer(new JsonDebeziumDeserializationSchema())
    .debeziumProperties(
        PropertiesUtil.debeziumProperties()
            .setProperty("binary.handling.mode", "hex")
            .build()
    )
    .build();

注意事项

1. 性能考虑：二进制到字符串的转换会增加处理开销
2. 主键唯一性：确保转换后的主键仍能保持唯一性
3. 下游兼容性：确保下游系统能处理转换后的主键格式
4. Flink CDC 版本：较新版本对二进制类型支持更好，考虑升级

如果以上方案仍不能解决问题，建议查看具体的错误日志并提供更多细节以便进一步分析。