databend debezium server support auto schema evolution

背景

Debezium Server Databend 是一个基于 Debezium Engine 自研的轻量级 CDC 项目，用于实时捕获数据库更改并将其作为事件流传递最终将数据写入目标数据库 Databend。它提供了一种简单的方式来监视和捕获关系型数据库的变化，并支持将这些变化转换为可消费事件。使用 Debezium server databend 实现 CDC 无须依赖大型的 Data Infra 比如 Flink, Kafka, Spark 等，只需一个启动脚本即可开启实时数据同步。

CDC 过程中的 Schema 变更处理是上游数据库中十分常见的用户场景，也是数据同步框架实现的难点。针对该场景，Databend-debezium-server 0.3.0 引入了 Auto Schema Evolution 的能力，在每一批次的数据中协调并控制作业拓扑中的 schema 变更事件处理。

实现过程

Debezium Server Databend 实现 Schema Auto Evolution 功能的原理大致是：

在配置文件中新增 debezium.sink.databend.schema.evolution 的配置，默认为 false 不开启该功能。

当上游数据源发生 schema 变更时，先将流水线中已经读出的的数据全部刷出以保证进入数据流的这一批 schema 的一致性：

然后先将该类 schemachangekey 事件暂存到 schemaEvolutionEvents 的 ArrayList 中。

List<DatabendChangeEvent> schemaEvolutionEvents = new ArrayList<>();
  for (DatabendChangeEvent event : events) {
   if (DatabendUtil.isSchemaChanged(event.schema()) && isSchemaEvolutionEnabled) {
             schemaEvolutionEvents.add(event);
         }
    }

先将上面刷出的数据执行写入操作，写入这批数据之后且解析 schema 变更的事件之前的时间里不会有新的数据进来。数据处理完后再去解析 schema change events，如果事件类型属于 DDL 并且为 alter table 语句，就对目标 database.table 执行该 DDL。

// handle schema evolution
      try {
          schemaEvolution(table, schemaEvolutionEvents);
      } catch (Exception e) {
          throw new RuntimeException(e.getMessage());
      }

public void schemaEvolution(RelationalTable table, List<DatabendChangeEvent> events) {
     for (DatabendChangeEvent event : events) {
         Map<String, Object> values = event.valueAsMap();
         for (Map.Entry<String, Object> entry : values.entrySet()) {
             if (entry.getKey().contains("ddl") && entry.getValue().toString().toLowerCase().contains("alter table")) {
                 String tableName = getFirstWordAfterAlterTable(entry.getValue().toString());
                 String ddlSql = replaceFirstWordAfterTable(entry.getValue().toString(), table.databaseName + "." + tableName);
                 try (PreparedStatement statement = connection.prepareStatement(ddlSql)) {
                     System.out.println(ddlSql);
                     statement.execute(ddlSql);
                 } catch (SQLException e) {
                     throw new RuntimeException(e.getMessage());
                 }
             }
         }
     }
 }

当 schema 变更事件处理成功后，会继续新的数据同步流程。

基本的处理流程如下图所示：

实践&演示

Debezium Server Databend

Clone 项目: git clone `https://github.com/databendcloud/debezium-server-databend.git`

从项目根目录开始:

构建和打包 debezium server: mvn -Passembly -Dmaven.test.skip package
构建完成后，解压服务器分发包: unzip debezium-server-databend-dist/target/debezium-server-databend-dist*.zip -d databendDist
进入解压后的文件夹: cd databendDist
创建 application.properties 文件并修改: nano conf/application.properties，将下面的 application.properties 拷贝进去，根据用户实际情况修改相应的配置。

debezium.sink.type=databend
debezium.sink.databend.upsert=true
debezium.sink.databend.upsert-keep-deletes=false
debezium.sink.databend.database.databaseName=debezium
debezium.sink.databend.database.url=jdbc:databend://tnf34b0rm--xxxxxx.default.databend.cn:443
debezium.sink.databend.database.username=cloudapp
debezium.sink.databend.database.password=password
debezium.sink.databend.database.primaryKey=id
debezium.sink.databend.database.tableName=products
debezium.sink.databend.database.param.ssl=true
debezium.sink.databend.schema.evolution=true // Enable Auto Schema Evolution

# enable event schemas
debezium.format.value.schemas.enable=true
debezium.format.key.schemas.enable=true
debezium.format.value=json
debezium.format.key=json

# mysql source
debezium.source.connector.class=io.debezium.connector.mysql.MySqlConnector
debezium.source.offset.storage.file.filename=data/offsets.dat
debezium.source.offset.flush.interval.ms=60000

debezium.source.database.hostname=127.0.0.1
debezium.source.database.port=3306
debezium.source.database.user=root
debezium.source.database.password=123456
debezium.source.database.dbname=mydb
debezium.source.database.server.name=from_mysql
debezium.source.include.schema.changes=false
debezium.source.table.include.list=mydb.products
# debezium.source.database.ssl.mode=required
# Run without Kafka, use local file to store checkpoints
debezium.source.database.history=io.debezium.relational.history.FileDatabaseHistory
debezium.source.database.history.file.filename=data/status.dat
# do event flattening. unwrap message!
debezium.transforms=unwrap
debezium.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
debezium.transforms.unwrap.delete.handling.mode=rewrite
debezium.transforms.unwrap.drop.tombstones=true

# ############ SET LOG LEVELS ############
quarkus.log.level=INFO
# Ignore messages below warning level from Jetty, because it's a bit verbose
quarkus.log.category."org.eclipse.jetty".level=WARN

使用提供的脚本运行服务: bash run.sh
Debezium Server with Databend 将会启动

Mysql 中准备表和数据

创建数据库 mydb 和表 products，并插入数据:

CREATE DATABASE mydb;
USE mydb;

CREATE TABLE products (id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,name VARCHAR(255) NOT NULL,description VARCHAR(512));
ALTER TABLE products AUTO_INCREMENT = 10;

INSERT INTO products VALUES (default,"scooter","Small 2-wheel scooter"),
(default,"car battery","12V car battery"),
(default,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3"),
(default,"hammer","12oz carpenter's hammer"),
(default,"hammer","14oz carpenter's hammer"),
(default,"hammer","16oz carpenter's hammer"),
(default,"rocks","box of assorted rocks"),
(default,"jacket","water resistent black wind breaker"),
(default,"cloud","test for databend"),
(default,"spare tire","24 inch spare tire");