Flink CDC用的stream api 自定义反序列化器显示类型转换问题？

在使用Flink CDC的DataStream API时，如果自定义反序列化器（如JsonDebeziumDeserializationSchema）出现类型转换问题，通常是因为Debezium在解析数据时对某些数据类型（如数值类型）采用了不同的转换方式。以下是解决该问题的详细步骤和说明：

1. 问题原因

Debezium在解析MySQL Binlog中的数据时，会根据配置将不同类型的字段转换为特定的格式。例如： - 数值类型（如BIGINT UNSIGNED或DECIMAL）可能会被转换为字符串或其他非预期格式。 - 这种行为可能导致下游处理逻辑无法正确解析数据。

2. 解决方案

通过配置Debezium的属性来控制数值类型的解析方式，确保数据以预期的格式传递到下游。

步骤 1：设置Debezium属性

在构建MySqlSource时，可以通过debeziumProperties方法传递自定义的Debezium配置。以下是一些常用的配置项：

Properties properties = new Properties();
properties.setProperty("bigint.unsigned.handling.mode", "long"); // 将无符号BIGINT解析为Long
properties.setProperty("decimal.handling.mode", "double");       // 将DECIMAL解析为Double

MySqlSource.<String>builder()
    .hostname(config.getHostname())
    .port(config.getPort())
    .databaseList(config.getDatabaseList())
    .username(config.getUsername())
    .password(config.getPassword())
    .deserializer(new JsonDebeziumDeserializationSchema()) // 使用JSON反序列化器
    .debeziumProperties(properties) // 设置Debezium属性
    .build();

• bigint.unsigned.handling.mode：控制无符号BIGINT的解析方式，可选值包括long（默认）、precise（精确值）等。
• decimal.handling.mode：控制DECIMAL类型的解析方式，可选值包括double（浮点数）、string（字符串）等。

步骤 2：验证数据格式

在下游处理逻辑中，验证数据是否以预期格式传递。例如： - 如果BIGINT UNSIGNED字段被正确解析为Long，则可以直接进行数值计算。 - 如果DECIMAL字段被解析为Double，则需要注意精度损失问题。

3. 注意事项

• 数据精度：将DECIMAL解析为Double可能会导致精度损失。如果需要高精度计算，建议将其解析为BigDecimal，并在下游逻辑中显式处理。
• 兼容性：确保Debezium版本与Flink CDC版本兼容，避免因版本不匹配导致的解析问题。
• 调试建议：在开发环境中打印反序列化后的数据，确认字段类型和值是否符合预期。

4. 示例代码

以下是一个完整的示例代码，展示如何配置Debezium属性并使用自定义反序列化器：

import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;

public class FlinkCDCExample {
    public static void main(String[] args) throws Exception {
        Properties properties = new Properties();
        properties.setProperty("bigint.unsigned.handling.mode", "long");
        properties.setProperty("decimal.handling.mode", "double");

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname("localhost")
            .port(3306)
            .databaseList("test_db")
            .username("root")
            .password("password")
            .deserializer(new JsonDebeziumDeserializationSchema())
            .debeziumProperties(properties)
            .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(mySqlSource).print();

        env.execute("Flink CDC Example");
    }
}

5. 总结

通过合理配置Debezium属性，可以有效解决Flink CDC中自定义反序列化器的类型转换问题。建议根据实际业务需求选择合适的解析方式，并在开发过程中进行充分测试以确保数据一致性。