Debezium捕获MySQL CDC数据类型不一致？自定义Converter与Schema Registry演进

大家好，今天我们来深入探讨在使用Debezium捕获MySQL CDC（Change Data Capture）数据时，可能遇到的数据类型不一致问题，以及如何利用自定义Converter和Schema Registry来解决和优化这些问题。

1. CDC数据类型不一致的常见场景

在使用Debezium监听MySQL数据库的变更时，我们经常会遇到以下几种数据类型不一致的情况：

MySQL特有类型到通用类型的映射问题： 例如，MySQL的ENUM或SET类型，在Debezium默认的配置下可能被转换为String，但下游系统可能更需要数值类型的枚举值或者Set集合的字符串数组。
精度丢失： MySQL的DECIMAL类型如果精度很高，在转换为JSON或Avro时可能出现精度丢失，尤其是在下游系统使用float或double类型接收的情况下。
时区问题： MySQL的TIMESTAMP类型存储的是UTC时间，但在Debezium处理过程中，可能受到服务器时区的影响，导致时间表示不一致。
BLOB/TEXT类型处理： 默认情况下，Debezium可能不会完整地捕获BLOB或TEXT类型的数据，或者以不方便下游系统处理的方式传递。
NULL值处理不一致： MySQL允许对某些数据类型（如TIMESTAMP）插入NULL值，但在Debezium处理后，可能被转换为其他默认值，导致数据含义发生变化。

举例：

假设我们有一个MySQL表products：

CREATE TABLE products (
    id INT PRIMARY KEY,
    name VARCHAR(255),
    price DECIMAL(10, 2),
    status ENUM('active', 'inactive', 'pending')
);

INSERT INTO products (id, name, price, status) VALUES (1, 'Sample Product', 99.99, 'active');

如果使用默认配置的Debezium Connector来捕获products表的变更，那么status字段的值可能被转换为字符串"active"，而下游系统更希望接收到数值类型的枚举值（例如，'active' = 0, 'inactive' = 1, 'pending' = 2）。 price字段可能因为下游系统使用了float类型而导致精度丢失。

2. Debezium Converter的作用

Debezium Converter负责将Debezium捕获到的数据从一种格式转换为另一种格式。更具体地说，Converter负责将MySQL数据类型转换为Debezium Schema中定义的数据类型，然后再将Debezium Schema中的数据类型转换为目标消息格式（例如JSON、Avro）。

Debezium提供了多种内置的Converter，例如：

org.apache.kafka.connect.json.JsonConverter： 将数据转换为JSON格式。
io.confluent.connect.avro.AvroConverter： 将数据转换为Avro格式。
org.apache.kafka.connect.converters.ByteArrayConverter： 将数据转换为字节数组。

这些内置的Converter通常可以满足大部分需求，但对于一些特定的数据类型转换或自定义逻辑，我们需要编写自定义的Converter。

3. 自定义Converter的实现

要实现自定义的Converter，我们需要创建一个继承自org.apache.kafka.connect.converter.Converter的类，并实现其configure、fromConnectData和toConnectData方法。

configure(Map<String, ?> configs, boolean isKey)： 用于配置Converter，例如读取配置参数。isKey参数表示当前Converter是用于处理Key还是Value。
fromConnectData(String topic, Schema schema, Object value)： 将Kafka Connect的数据（ConnectData）转换为字节数组，通常用于Producer端。
toConnectData(String topic, Headers headers, byte[] value)： 将字节数组转换为Kafka Connect的数据（ConnectData），通常用于Consumer端。 Headers参数允许你访问Kafka记录的header信息。

示例：自定义ENUM类型Converter

假设我们想要将MySQL的ENUM类型转换为数值类型的枚举值。我们可以创建一个名为EnumConverter的自定义Converter：

import org.apache.kafka.connect.converter.Converter;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.errors.DataException;
import org.apache.kafka.connect.header.Headers;

import java.util.Map;

public class EnumConverter implements Converter {

    private Map<String, Map<String, Integer>> enumMappings; // Enum name to value mapping
    private boolean isKey;

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        this.isKey = isKey;
        // Load enum mappings from configuration.  For example:
        // enumMappings = loadEnumMappings(configs.get("enum.mappings"));
        // For simplicity, let's hardcode a mapping for our example:
        enumMappings = Map.of(
                "products.status", Map.of("active", 0, "inactive", 1, "pending", 2)
        );
    }

    @Override
    public byte[] fromConnectData(String topic, Schema schema, Object value) {
        if (value == null) {
            return null;
        }

        if (!(value instanceof String)) {
            throw new DataException("Expected String type for ENUM, but found: " + value.getClass());
        }

        String enumValue = (String) value;
        String fieldName = schema.name(); // Assuming schema name is the field name
        String tableName = topic.substring(topic.lastIndexOf(".") + 1); // Extract table name from topic

        String fullFieldName = tableName + "." + fieldName;

        if (!enumMappings.containsKey(fullFieldName)) {
            throw new DataException("No mapping found for enum field: " + fullFieldName);
        }

        Map<String, Integer> mapping = enumMappings.get(fullFieldName);
        if (!mapping.containsKey(enumValue)) {
            throw new DataException("Invalid enum value: " + enumValue + " for field: " + fullFieldName);
        }

        Integer intValue = mapping.get(enumValue);

        // Convert the Integer to byte array (e.g., using ByteBuffer)
        // This part depends on your desired output format.
        // For example, if using Avro, you'd create an Avro record here.
        // For simplicity, we just return the Integer as a String byte array:
        return String.valueOf(intValue).getBytes();
    }

    @Override
    public SchemaAndValue toConnectData(String topic, Headers headers, byte[] value) {
        if (value == null) {
            return SchemaAndValue.NULL;
        }

        String stringValue = new String(value); // Assuming the value is a String
        Integer intValue;
        try {
             intValue = Integer.parseInt(stringValue);
        } catch (NumberFormatException e) {
            throw new DataException("Expected an integer value, but found: " + stringValue, e);
        }

        String fieldName = extractFieldNameFromTopic(topic); // Implement this method
        String tableName = extractTableNameFromTopic(topic); // Implement this method

        String fullFieldName = tableName + "." + fieldName;

        // Reverse lookup: Find the enum value based on the integer value.
        String enumValue = null;
        for (Map.Entry<String, Map<String, Integer>> entry : enumMappings.entrySet()) {
            if (entry.getKey().equals(fullFieldName)) {
                for (Map.Entry<String, Integer> enumEntry : entry.getValue().entrySet()) {
                    if (enumEntry.getValue().equals(intValue)) {
                        enumValue = enumEntry.getKey();
                        break;
                    }
                }
                break;
            }
        }

        if (enumValue == null) {
            throw new DataException("No enum value found for integer value: " + intValue + " and field: " + fullFieldName);
        }

        // Create a schema for the enum string value
        Schema enumSchema = SchemaBuilder.string().name(fieldName).optional().build();

        return new SchemaAndValue(enumSchema, enumValue);
    }

    private String extractFieldNameFromTopic(String topic) {
        // Implement logic to extract the field name from the topic
        // For example, if the topic is "dbserver1.database1.products.status",
        // you might extract "status".
        // This implementation is just a placeholder and needs to be adjusted based on your topic naming convention.
        return "status"; // Placeholder
    }

    private String extractTableNameFromTopic(String topic) {
         // Implement logic to extract the table name from the topic
        // For example, if the topic is "dbserver1.database1.products.status",
        // you might extract "products".
        // This implementation is just a placeholder and needs to be adjusted based on your topic naming convention.
        return "products"; // Placeholder
    }

    @Override
    public void close() {
        // Clean up resources if needed
    }
}

重要说明：

这个示例代码只是一个框架。你需要根据你的实际需求来实现loadEnumMappings方法，从配置文件或数据库中加载枚举值的映射关系。
fromConnectData方法将ENUM字符串值转换为整数值的字节数组。你需要根据你的目标消息格式（例如Avro、JSON）来调整转换逻辑。如果使用Avro，你需要创建一个Avro record，并将整数值放入该record中。
toConnectData方法将整数值的字节数组转换回ENUM字符串值。同样，你需要根据你的实际需求来调整转换逻辑。
extractFieldNameFromTopic 和 extractTableNameFromTopic 需要你根据你的topic naming convention 实现。
错误处理至关重要。在Converter中，务必处理各种可能的异常情况，并抛出DataException，以便Kafka Connect能够正确地处理错误。

4. 配置Debezium使用自定义Converter

要让Debezium Connector使用自定义的Converter，你需要在Connector的配置中指定Converter的类名，并配置相关的参数。

例如，如果使用Kafka Connect的REST API来创建Connector，你可以这样配置：

{
  "name": "my-mysql-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "localhost",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "184054",
    "database.server.name": "my-app-connector",
    "database.include.list": "mydatabase",
    "table.include.list": "mydatabase.products",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "com.example.EnumConverter", // 使用自定义Converter
    "value.converter.enum.mappings": "products.status:active=0,inactive=1,pending=2", // 配置Converter参数
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState"
  }
}

关键点：

value.converter：指定自定义Converter的完整类名。
value.converter.enum.mappings：配置Converter的参数。参数的名称必须以value.converter.开头。
确保自定义Converter的jar文件在Kafka Connect的classpath中。你可以将jar文件放在Kafka Connect的plugin.path目录下。

5. Schema Registry的引入和演进

当使用Avro作为消息格式时，Schema Registry可以帮助我们管理和共享Avro schema。 Schema Registry是一个集中式的存储库，用于存储Avro schema，并为Producer和Consumer提供schema注册和查找服务。

Schema Registry的优势：

schema演进： Schema Registry支持schema演进，允许我们在不中断现有系统的情况下修改schema。
schema复用： 不同的Producer和Consumer可以共享同一个schema，避免了schema重复定义的问题。
数据验证： Schema Registry可以对Producer发送的数据进行验证，确保数据符合schema的定义。
降低存储空间： 消息中只需要包含schema的ID，而不是完整的schema定义，从而降低了存储空间。

Debezium与Schema Registry集成：

Debezium可以与Schema Registry无缝集成。当Debezium捕获到MySQL的变更时，它会将数据的Avro schema注册到Schema Registry中，并将schema的ID包含在消息中。 Consumer可以使用schema ID从Schema Registry中获取schema，并对消息进行反序列化。

配置Debezium使用Schema Registry：

要让Debezium Connector使用Schema Registry，你需要在Connector的配置中指定Schema Registry的地址。

例如：

{
  "name": "my-mysql-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "localhost",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "184054",
    "database.server.name": "my-app-connector",
    "database.include.list": "mydatabase",
    "table.include.list": "mydatabase.products",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://localhost:8081", // Schema Registry 地址
    "value.converter.schema.registry.url": "http://localhost:8081", // Schema Registry 地址
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState"
  }
}

Schema Registry演进策略：

Schema Registry支持多种schema演进策略，例如：

BACKWARD： 新的schema可以读取使用旧schema编写的数据。
FORWARD： 旧的schema可以读取使用新的schema编写的数据。
FULL： 新的schema和旧的schema可以互相读取对方编写的数据。
NONE： 不允许schema演进。

选择合适的schema演进策略取决于你的业务需求。一般来说，BACKWARD策略是最常用的，因为它允许我们在不中断现有系统的情况下添加新的字段。

6. 结合自定义Converter和Schema Registry

我们可以将自定义Converter和Schema Registry结合使用，以实现更灵活的数据类型转换和schema管理。

例如，我们可以创建一个自定义的Avro Converter，该Converter在将数据转换为Avro格式之前，先对数据进行一些自定义的转换，例如将ENUM类型转换为数值类型的枚举值。

import io.confluent.connect.avro.AvroConverter;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaAndValue;
import org.apache.kafka.connect.header.Headers;

import java.util.Map;

public class CustomAvroConverter extends AvroConverter {

    private Map<String, Map<String, Integer>> enumMappings;

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        super.configure(configs, isKey);
        // Load enum mappings from configuration.  For example:
        // enumMappings = loadEnumMappings(configs.get("enum.mappings"));
        enumMappings = Map.of(
                "products.status", Map.of("active", 0, "inactive", 1, "pending", 2)
        );
    }

    @Override
    public SchemaAndValue toConnectData(String topic, Headers headers, byte[] value) {
        // Call the super class to convert the byte array to ConnectData
        SchemaAndValue schemaAndValue = super.toConnectData(topic, headers, value);

        // Apply custom data transformation here
        Schema schema = schemaAndValue.schema();
        Object valueObject = schemaAndValue.value();

        // Example: Convert enum field to integer
        if (schema != null && valueObject != null) {
            String fieldName = extractFieldNameFromTopic(topic); // Implement this method
            String tableName = extractTableNameFromTopic(topic); // Implement this method
            String fullFieldName = tableName + "." + fieldName;

            if (enumMappings.containsKey(fullFieldName) && schema.type().getName().equals("string")) {
                String enumValue = (String) valueObject;
                Map<String, Integer> mapping = enumMappings.get(fullFieldName);
                Integer intValue = mapping.get(enumValue);

                // Create a new Schema with Integer type
                Schema intSchema = org.apache.kafka.connect.data.SchemaBuilder.int32().optional().build();

                return new SchemaAndValue(intSchema, intValue);
            }
        }

        return schemaAndValue;
    }

    private String extractFieldNameFromTopic(String topic) {
        // Implement logic to extract the field name from the topic
        return "status"; // Placeholder
    }

    private String extractTableNameFromTopic(String topic) {
        // Implement logic to extract the table name from the topic
        return "products"; // Placeholder
    }

}

在这个示例中，我们继承了AvroConverter，并重写了toConnectData方法。在toConnectData方法中，我们首先调用父类的toConnectData方法，将字节数组转换为ConnectData。然后，我们对ConnectData中的数据进行自定义的转换，例如将ENUM类型转换为数值类型的枚举值。最后，我们返回一个新的SchemaAndValue对象，其中包含了转换后的schema和数据。

7. 总结：灵活应对数据类型挑战

总而言之，Debezium的Converter机制为我们提供了强大的数据类型转换能力。通过自定义Converter，我们可以灵活地处理MySQL的特有数据类型，并将其转换为下游系统更容易处理的格式。结合Schema Registry，我们可以实现schema的演进和复用，提高系统的可维护性和扩展性。

Debezium捕获MySQL CDC数据类型不一致？自定义Converter与Schema Registry演进

发表回复 取消回复

发表回复取消回复