c#-Apache Paimon 使用 Kafka CDC 获取数据

a.依赖准备

flink-sql-connector-kafka-*.jar

b.支持的文件格式

Flink提供了几种Kafka CDC格式：Canal、Debezium、Ogg和Maxwell JSON。

如果Kafka的Topic中的消息是使用Change Data Capture（CDC）工具从另一个数据库捕获的change event，那么可以使用Paimon Kafka CDC，将INSERT、UPDATE、DELETE消息写入到paimon表中。

注意：

JSON源可能缺少一些信息。例如，Ogg和Maxwell格式标准不包含字段类型；当将JSON源写入Flink Kafka Sink时，它只会保留数据和行类型并删除其他信息。

通常，debezium-json包含“schema”字段，Paimon将从中检索数据类型。确保debezium json具有此字段，否则Paimon将使用“STRING”类型。

• 如果缺少字段类型，Paimon将默认使用“STRING”类型。
• 如果缺少数据库名称或表名，则无法进行数据库同步，但仍然可以进行表同步。
• 如果缺少主键，该作业可能会创建非主键表，可以在表synchronization中提交作业时设置主键。

c.同步表

在Flink DataStream作业中使用 KafkaSyncTableAction 或直接通过flink run，可以将Kafka的一个Topic中的一个或多个表同步到一个Paimon表中。

<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-0.7.0-incubating.jar \
    kafka_sync_table
    --warehouse <warehouse-path> \
    --database <database-name> \
    --table <table-name> \
    [--partition_keys <partition_keys>] \
    [--primary_keys <primary-keys>] \
    [--type_mapping to-string] \
    [--computed_column <'column-name=expr-name(args[, ...])'> [--computed_column ...]] \
    [--kafka_conf <kafka-source-conf> [--kafka_conf <kafka-source-conf> ...]] \
    [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]] \
    [--table_conf <paimon-table-sink-conf> [--table_conf <paimon-table-sink-conf> ...]]

如果指定的Paimon表不存在，此操作将自动创建该表，它的结构将从所有指定的Kafka的Topic的表中派生出来，它从Topic中获取最早的非DDL数据解析模式。

如果Paimon表已经存在，其模式将与所有指定的Kafka的Topic表的模式进行比较。

示例1：

<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-0.7.0-incubating.jar \
    kafka_sync_table \
    --warehouse hdfs:///path/to/warehouse \
    --database test_db \
    --table test_table \
    --partition_keys pt \
    --primary_keys pt,uid \
    --computed_column '_year=year(age)' \
    --kafka_conf properties.bootstrap.servers=127.0.0.1:9020 \
    --kafka_conf topic=order \
    --kafka_conf properties.group.id=123456 \
    --kafka_conf value.format=canal-json \
    --catalog_conf metastore=hive \
    --catalog_conf uri=thrift://hive-metastore:9083 \
    --table_conf bucket=4 \
    --table_conf changelog-producer=input \
    --table_conf sink.parallelism=4

如果启动同步作业时kafka的Topic中不包含消息，则必须在提交作业之前手动创建表，只能定义分区键和主键，剩下的列将由同步作业添加。

注意：在这种情况下，不应该使用-partition_keys或-primary_keys，因为这些键是在创建表时定义的，不能修改。此外，如果指定了计算列，还应该定义用于计算列的所有参数列。

示例2：如果想同步具有主键“id INT”的表，并且要计算分区键“part=date_format(create_time,yyyy-MM-dd)”，可以先创建这样的表（其他列可以省略）

CREATE TABLE test_db.test_table (
    id INT,                 -- primary key
    create_time TIMESTAMP,  -- the argument of computed column part
    part STRING,            -- partition key
    PRIMARY KEY (id, part) NOT ENFORCED
) PARTITIONED BY (part);

启动同步作业

<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-0.7.0-incubating.jar \
    kafka_sync_table \
    --warehouse hdfs:///path/to/warehouse \
    --database test_db \
    --table test_table \
    --computed_column 'part=date_format(create_time,yyyy-MM-dd)' \
    ... (other conf)

d.同步数据库

通过在Flink DataStream作业中使用KafkaSyncDatabaseAction或直接通过flink run，可以将多个Topic或一个Topic同步到一个Paimon数据库中。

<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-0.7.0-incubating.jar \
    kafka_sync_database
    --warehouse <warehouse-path> \
    --database <database-name> \
    [--table_prefix <paimon-table-prefix>] \
    [--table_suffix <paimon-table-suffix>] \
    [--including_tables <table-name|name-regular-expr>] \
    [--excluding_tables <table-name|name-regular-expr>] \
    [--type_mapping to-string] \
    [--kafka_conf <kafka-source-conf> [--kafka_conf <kafka-source-conf> ...]] \
    [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]] \
    [--table_conf <paimon-table-sink-conf> [--table_conf <paimon-table-sink-conf> ...]]

只有带有主键的表才会同步。

此操作将为所有表构建一个combined sink，对于要同步的每个Kafka Topic的表，如果相应的Paimon表不存在，此操作将自动创建该表，其模式将从所有指定的Kafka Topic的表中派生。

如果Paimon表已经存在，并且其模式与从Kafka记录中解析的模式不同，则此操作将尝试模式演变。

示例：从一个Kafka Topic同步到Paimon数据库。

<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-0.7.0-incubating.jar \
    kafka_sync_database \
    --warehouse hdfs:///path/to/warehouse \
    --database test_db \
    --kafka_conf properties.bootstrap.servers=127.0.0.1:9020 \
    --kafka_conf topic=order \
    --kafka_conf properties.group.id=123456 \
    --kafka_conf value.format=canal-json \
    --catalog_conf metastore=hive \
    --catalog_conf uri=thrift://hive-metastore:9083 \
    --table_conf bucket=4 \
    --table_conf changelog-producer=input \
    --table_conf sink.parallelism=4

从多个Kafka Topic同步到Paimon数据库。

<FLINK_HOME>/bin/flink run \
    /path/to/paimon-flink-action-0.7.0-incubating.jar \
    kafka_sync_database \
    --warehouse hdfs:///path/to/warehouse \
    --database test_db \
    --kafka_conf properties.bootstrap.servers=127.0.0.1:9020 \
    --kafka_conf topic=order\;logistic_order\;user \
    --kafka_conf properties.group.id=123456 \
    --kafka_conf value.format=canal-json \
    --catalog_conf metastore=hive \
    --catalog_conf uri=thrift://hive-metastore:9083 \
    --table_conf bucket=4 \
    --table_conf changelog-producer=input \
    --table_conf sink.parallelism=4