《Debezium系列》Debezium连接器使用Avro序列化

Bertha 。 2023-05-22 15:25 57阅读 0赞

官方文章参考：

[https://debezium.io/documentation/reference/1.1/configuration/avro.html][https_debezium.io_documentation_reference_1.1_configuration_avro.html]

书籍参考： kafka权威指南

# 1. 起因 #

对于CDC产生的事件信息来说，debezium给出的消息内容过多，其中对于元数据的结构描述长并且每条消息都需要参夹发送。刚好官方文档提供除了String、Json以外的一种序列化工具：Avro。

并使用Confluent提供的Schema-Registry进行Avro的相关配置。

它，短小精悍；它，二进制大小小；它，schema修改更新元数据，不会报错；它：我要打十个。

# 2. Avro简介 #

Avro为Apache项目，是一个数据序列化系统。序列化格式。（官网展示相关特性如下）

> Avro provides:
> 
>  *  Rich data structures.
>  *  A compact, fast, binary data format.
>  *  A container file, to store persistent data.
>  *  Remote procedure call (RPC).
>  *  Simple integration with dynamic languages. Code generation is not required to read or write data files nor to use or implement RPC protocols. Code generation as an optional optimization, only worth implementing for statically typed languages.
> 
>

权威指南中有：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70][]

其中Avro重要的是，就是写消息的应用程序使用了新的schema（例如更改了字段类型或者新增字段），负责读的应用程序可以继续处理消息而不用做改动。

例如有一个类：

public class Customer {
        private int id;
        private String name;
        private String faxNumber;
    }
    
    // 某天它变成了
    
    public class Customer {
        private int id;
        private String name;
        private String email;
    }

在应用程序升级之前，它们会调用类似 getName（）、getId（） 和 getFaxNumber（） 这样的方法。如果碰到使用新 schema 构建的消息，getName（） 和 getId（） 方法仍然能够正常返回，但 getFaxNumber（） 方法会返回 null，因为消息里不包含传真号码。

在应用程序升级之后，getEmail（） 方法取代了 getFaxNumber（） 方法。如果碰到一个使用旧 schema 构建的消息，那么 getEmail（） 方法会返回 null，因为旧消息不包含邮件地址。

我们修改了消息的 schema，但并没有更新所有负责读取数据的应用程序，而这样仍然不会出现异常或阻断性错误，也不需要对现有数据进行大幅更新。

其中最重要的是：

它通过schema来定义数据，而schema是用json来描述的。数据会被序列化成二进制格式进行传输，Avro在读写文件的时候需要用到schema，schema一般会被内嵌到数据文件中。

简单示例：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 1][]

对于kafka来说，不可能在每条消息中嵌套schema消息，这个会增加数据大小，传输的时候增大开销。

所以这个时候就需要一个中间组件，注册表（schema-registry），在生产消息的时候相关的schema存储在注册表中并给与一个id可以寻找到该schema（该注册表不属于kafka）。然后在读取该消息的时候去注册表拿取schema信息。

所以这就很适合kafka这样的消息系统。

# 3. 在Kafka中的Avro简单例子 #

道理我都懂，要不实践一下？

所以通过下面这个生产者例子了解下Avro的实际使用。

import org.apache.avro.Schema;
    import org.apache.avro.generic.GenericData;
    import org.apache.avro.generic.GenericRecord;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.Producer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import java.util.Properties;
    
    public class AvroProducer {
        public static void main(String[] args) {
            Properties pros = new Properties();
            pros.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            // 选择Confluent Avro序列化
            pros.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroSerializer");
            pros.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroSerializer");
            // schema-registry地址
            pros.put("schema.registry.url", "http://schemaUrl:8081");
            // schema info
            String schemaString = "{\"namespace\":\"customerManagement.avro\",\"type\":\"record\"," +
                    "\"name\" : \"Customer\", \"fields\" : [{\"name\":\"id\",\"type\":\"string\"}," +
                    "{\"name\": \"name\", \"fields\" : \"string\"}" +
                    "{\"name\" : \"email\", \"type\" : [\"null\",\"string\"], \"default\" : \"null\"}]}";
            Producer<String, GenericRecord> producer = new KafkaProducer<String, GenericRecord>(pros);
            // 构建Schema
            Schema.Parser parser = new Schema.Parser();
            Schema schema = parser.parse(schemaString);
            // 模拟消息发送
            for (int i = 0; i < 10; i++) {
                GenericRecord customer = new GenericData.Record(schema);
                String name = "customer" + i;
                // 此处会进行映射
                customer.put("id", name);
                customer.put("name", "MyName");
                customer.put("email", "name@name.com");
                // 指定发送的topic、key、value
                ProducerRecord<String, GenericRecord> data = new ProducerRecord<String, GenericRecord>
                        ("customerContacts", name, customer);
                // 发送
                producer.send(data);
            }
        }
    }

# 4. 为Debezium连接器配置Confluent的Avro Schema Registry #

其实这个标题起的不太好，因为Confluent的Scheam Registry并不只是Avro专用，也支持其他序列化，它只是一个注册表而已。

官方传送---[https://docs.confluent.io/current/schema-registry/index.html][https_docs.confluent.io_current_schema-registry_index.html]

其Schema的架构如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 2][]

## 4.1 安装Confluent配置Schema ##

例如在centos下安装：

[https://docs.confluent.io/5.4.1/installation/installing\_cp/rhel-centos.html\#systemd-rhel-centos-install][https_docs.confluent.io_5.4.1_installation_installing_cp_rhel-centos.html_systemd-rhel-centos-install]

这里推荐安装社区版本就够用了。

修改该目录下的配置文件：

![20200430150054169.png][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 3][]

启动schema

[root@host01 schema-registry]# schema-registry-start schema-registry.properties

## 4.2. Connect全局配置或者提交connector时进行配置 ##

connect.properties

key.converter=io.confluent.connect.avro.AvroConverter
    key.converter.schema.registry.url=http://localhost:8081
    value.converter=io.confluent.connect.avro.AvroConverter
    value.converter.schema.registry.url=http://localhost:8081
    
    # 对于debezium来说，此处注意全部配置avro序列化时，不建议配置为avro
    # 需要配置kafka内部转换为json
    
    internal.key.converter=org.apache.kafka.connect.json.JsonConverter
    internal.value.converter=org.apache.kafka.connect.json.JsonConverter
    internal.key.converter.schemas.enable=false
    internal.value.converter.schemas.enable=false
    
    # 此处需要配置confluent的jar包目录，例如/usr/local/tools/confluent-5.4.1/share/java下面。
    # 若是不需要那么多的jar包，也可以去官网下载单独的包
    plugin=/usr/local/tools/confluent-5.4.1/share/java

Confluent的Avro序列化相关jar包：

[https://www.confluent.io/hub/confluentinc/kafka-connect-avro-converter][https_www.confluent.io_hub_confluentinc_kafka-connect-avro-converter]

下载完解压到connect指定的plugin目录下。

关于内部转换的官方解释：

![20200430152350211.png][]

或者进行connector提交时进行添加如上属性

"key.converter"="io.confluent.connect.avro.AvroConverter"
    "key.converter.schema.registry.url"="http://localhost:8081"
    "value.converter"="io.confluent.connect.avro.AvroConverter"
    "value.converter.schema.registry.url"="http://localhost:8081"

启动connect-distributed

或者

提交connector

若无Exception抛出则配置成功。

# 5. 遇到的问题 #

### 1. 进行Avro序列化提交的时候无法找到相关类 ###

我将jar包放进plugin文件夹后，启动各种connector和反复启动connect-distributed。

都没有将其包引入，一直NoClassDefFoundError，我都快怀疑人生了。

java.lang.NoClassDefFoundError: io/confluent/connect/avro/AvroConverterConfig

这种可能是jvm找不到该包classpath地址，导致未加载进去。执行下面语句解决。

export CLASSPATH=/usr/local/tools/kafka/kafka_2.12-2.4.0/plugin/*

### 2. 消费的数据不正确，例如Decimal格式数据未完全转换过来 ###

插入数据为：

INSERT INTO debeziumdb.data_type_test
    (id, varchar_data, bigint_data, float_data, double_date, decimal_data, date_data, time_data, datetime_data, timestamp_data)
    VALUES(1, 'www', 1, 1.10000002, 1.11, 5000.36, '2020-02-02', '12:12:10', '2020-02-02 12:12:10.0', NULL);

使用Confluent的Avro序列化消费结果为：

{"before": null, "after": {"id": 1, "varchar_data": "www", "bigint_data": 1, "float_data": 1.100000023841858, "double_date": 1.11, "decimal_data": "\u0007¡D", "date_data": 18294, "time_data": 43930000000, "datetime_data": 1580645530000, "timestamp_data": "1970-01-01T00:00:00Z"}, "source": {"version": "1.1.0.Final", "connector": "mysql", "name": "dbz.mysql", "ts_ms": 0, "snapshot": "last", "db": "debeziumdb", "table": "data_type_test", "server_id": 0, "gtid": null, "file": "MS-SJNKFHHAUOQV-bin.000003", "pos": 16244421, "row": 0, "thread": null, "query": null}, "op": "c", "ts_ms": 1588060753486, "transaction": null}

配置连接器属性：

"decimal.handling.mode": "string"
    
    或者
    
    "decimal.handling.mode": "double"

其中double类型保存两位精度。

# 6. 相关想法 #

大家都知道kafka是一个分布式架构，对于大量的请求可以分发缓解压力。

但是对于schema来说，假如我只启动了一个schema，之后的所有生产者都会将其schema注册到注册表中，如果压力过大会不会崩掉。

如果我启动多个schema，配置如何配置，是否会自动负载均衡？

这些都有待考究（可能Confluent已经考虑好了，我没发现而已）。

若是了解清楚了，会再出一篇说明。

[https_debezium.io_documentation_reference_1.1_configuration_avro.html]: https://debezium.io/documentation/reference/1.1/configuration/avro.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70]: /images/20230521/9cf4508a26d94dbb9abe4ed628a351ac.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 1]: /images/20230521/89c244c0ab304824817dc3aea06ca03f.png
[https_docs.confluent.io_current_schema-registry_index.html]: https://docs.confluent.io/current/schema-registry/index.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 2]: /images/20230521/6d123d6dd3b44556a08fd0b8e2939dd0.png
[https_docs.confluent.io_5.4.1_installation_installing_cp_rhel-centos.html_systemd-rhel-centos-install]: https://docs.confluent.io/5.4.1/installation/installing_cp/rhel-centos.html#systemd-rhel-centos-install
[20200430150054169.png]: /images/20230521/092d7fa6afde46a696eca4f6ba0fe6f4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTQ2MTk5Mg_size_16_color_FFFFFF_t_70 3]: /images/20230521/918c27c12686455d8f29d53b1a6f1a07.png
[https_www.confluent.io_hub_confluentinc_kafka-connect-avro-converter]: https://www.confluent.io/hub/confluentinc/kafka-connect-avro-converter
[20200430152350211.png]: /images/20230521/2bfb2498d1284350833bbeb2a8256aa3.png