第 13 章 开发 Debezium 自定义数据类型转换器

Debezium 更改事件记录中的每个字段代表源表或数据收集中的字段或列。当连接器向 Kafka 发送更改事件记录时，它会将源中的每个字段的数据类型转换为 Kafka Connect 模式类型。列值同样转换为与目的地字段的 schema 类型匹配。对于每个连接器，默认映射指定连接器如何转换每个数据类型。这些默认映射在每种连接器的数据类型文档中描述。

虽然默认映射通常就足够了，但对于某些应用程序，您可能需要应用备用映射。例如，如果默认映射导出自 UNIX epoch 以来的毫秒格式，您可能需要自定义映射，但下游应用只能以格式化的字符串形式消耗列值。您可以通过开发和部署自定义转换器来自定义数据类型映射。您可以将自定义转换器配置为对某个类型的所有列执行操作，或者您可以缩小其范围，使其仅适用于特定的表列。转换程序功能会截获与指定条件匹配的列的数据类型转换请求，然后执行指定的转换。转换程序会忽略与指定条件不匹配的列。

自定义转换器是实现 Debezium 服务供应商接口(SPI)的 Java 类。您可以通过在连接器配置中设置 converters 属性来启用和配置自定义转换器。converters 属性指定可用于连接器的转换器，可以包含进一步修改转换行为的子属性。

启动连接器后，连接器配置中启用的转换器将被实例化并添加到 registry 中。registry 将每个转换器与要处理的列或字段相关联。每当 Debezium 处理一个新的更改事件时，它会调用配置的转换器来转换其注册的列或字段。

public interface CustomConverter<S, F extends ConvertedField> {

    @FunctionalInterface
    interface Converter {  1
        Object convert(Object input);
    }

    public interface ConverterRegistration<S> { 2
        void register(S fieldSchema, Converter converter); 3
    }

    void configure(Properties props);

    void converterFor(F field, ConverterRegistration<S> registration); 4
}

public static class IsbnConverter implements CustomConverter<SchemaBuilder, RelationalColumn> {

    private SchemaBuilder isbnSchema;

    @Override
    public void configure(Properties props) {
        isbnSchema = SchemaBuilder.string().name(props.getProperty("schema.name"));
    }

    @Override
    public void converterFor(RelationalColumn column,
            ConverterRegistration<SchemaBuilder> registration) {

        if ("isbn".equals(column.typeName())) {
            registration.register(isbnSchema, x -> x.toString());
        }
    }
}

<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-api</artifactId>
    <version>${version.debezium}</version> 1
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>connect-api</artifactId>
    <version>${version.kafka}</version> 2
</dependency>