Debezium + Kafka-connect 实现Postgres实时同步Hologres

基于 Debezium + Kafka 的方案实现 PostgreSQL 到 Hologres 的实时数据同步，是一种高可靠性、高扩展性的解决方案。以下是详细的实现步骤：

1. 方案架构

Debezium：捕获 PostgreSQL 的变更数据（CDC），并将变更事件发送到 Kafka。

Kafka：作为消息队列，缓冲和分发变更数据。

Kafka 消费者：消费 Kafka 中的变更数据，并将其写入 Hologres。

Hologres：作为目标数据库，存储同步的数据。

2. 环境准备

2.1 组件版本

PostgreSQL：9.6 及以上（支持逻辑复制）。

Debezium：1.9 及以上。

Kafka：2.8 及以上。

Hologres：兼容 PostgreSQL 协议，支持 JDBC 连接。

2.2 配置 PostgreSQL

启用逻辑复制：

ALTER SYSTEM SET wal_level = logical;

创建复制槽：

# 创建复制槽（replication slot）：
SELECT pg_create_logical_replication_slot('debezium_slot', 'pgoutput');

# 查看复制插槽
SELECT * FROM pg_replication_slots;

创建发布：

# 创建发布(配置连接器可自动创建）
CREATE PUBLICATION debezium_pub FOR ALL TABLES;
CREATE PUBLICATION debezium_pub FOR TABLE my_table1, my_table2;# 删除发布
DROP PUBLICATION IF EXISTS debezium_pub;

确保 PostgreSQL 用户具有复制权限：

# 最好使用superuser
CREATE ROLE debezium_user WITH REPLICATION LOGIN PASSWORD 'your_password';

3. 配置 Debezium

3.1 下载 Debezium

链接下载 Debezium 的 PostgreSQL 连接器：
下载地址：https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/

wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/2.2.0.Final/debezium-connector-postgres-2.2.0.Final-plugin.tar.gz

3.2 添加插件到 Kafka Connect

手动复制插件 tar.gz 文件
将下载的 tar.gz 文件解压复制到 Kafka Connect 的插件目录中。 默认情况下，
插件目录为: /usr/share/java/kafka-connect-plugins/

# 创建插件目录（如果不存在）
sudo mkdir -p /usr/share/java/kafka-connect/# 复制 JAR 文件 
tar -zxvf /etc/kafka-connect/debezium-connector-postgres-2.2.0.Final-plugin.tar.gz -C /usr/share/java/kafka-connect/

使用 Kubernetes 挂载插件
如果你在 Kubernetes 环境中运行 Kafka Connect， 可以在 Deployment 文件中挂载插件目录：

volumes:- emptyDir: {}name: plugins-volume- emptyDir: {}name: config-volume​ volumeMounts:- name: plugins-volumemountPath: /usr/share/java/kafka-connect/- name: config-volumemountPath: /etc/kafka-connect/

Kafka-connect对应的Service、StatefulSets对象的yaml

apiVersion: v1
kind: Service
metadata:name: prod-kafka-connectnamespace: prod-core-jobs-shuguan
spec:ports:- port: 8083targetPort: 8083protocol: TCPname: httpselector:app: prod-kafka-connecttype: ClusterIP---
apiVersion: apps/v1
kind: StatefulSet
metadata:name: prod-kafka-connectnamespace: prod-core-jobs-shuguan
spec:serviceName: "prod-kafka-connect"replicas: 1selector:matchLabels:app: prod-kafka-connecttemplate:metadata:labels:app: prod-kafka-connectspec:volumes:- emptyDir: {}name: plugins-volume- emptyDir: {}name: config-volumeinitContainers:- name: extract-pluginsimage: pkg.geely.com/docker/busybox:1.31   # 初始化容器，作下载拷贝Debezium插件使用command: - sh- -c- |wget https://pkg.geely.com/artifactory/IMD-pypi-dev-hz/resource/uploadTemp/debezium-connector-postgres-2.2.0.Final-plugin.tar.gz -O /etc/kafka-connect/debezium-connector-postgres-2.2.0.Final-plugin.tar.gztar -zxvf /etc/kafka-connect/debezium-connector-postgres-2.2.0.Final-plugin.tar.gz -C /usr/share/java/kafka-connect/volumeMounts:- name: plugins-volumemountPath: /usr/share/java/kafka-connect/- name: config-volumemountPath: /etc/kafka-connect/containers:- name: prod-kafka-connectimage: pkg.geely.com/docker/confluentinc/cp-kafka-connect:7.2.0ports:- containerPort: 8083name: httpenv:- name: CONNECT_BOOTSTRAP_SERVERSvalue: "prod-kafka:9092"  # 替换为你的 Kafka 集群地址- name: CONNECT_GROUP_IDvalue: "connect-cluster"- name: CONNECT_KEY_CONVERTERvalue: "org.apache.kafka.connect.storage.StringConverter"- name: CONNECT_VALUE_CONVERTERvalue: "org.apache.kafka.connect.storage.StringConverter"- name: CONNECT_CONFIG_STORAGE_TOPICvalue: "connect-configs"- name: CONNECT_OFFSET_STORAGE_TOPICvalue: "connect-offsets"- name: CONNECT_STATUS_STORAGE_TOPICvalue: "connect-statuses"- name: CONNECT_REST_ADVERTISED_HOST_NAMEvalueFrom:fieldRef:fieldPath: status.podIP- name: CONNECT_REST_ADVERTISED_PORTvalue: "8083"volumeMounts:- name: plugins-volumemountPath: /usr/share/java/kafka-connect/- name: config-volumemountPath: /etc/kafka-connect/readinessProbe:httpGet:path: /connectorsport: httpinitialDelaySeconds: 30periodSeconds: 10livenessProbe:httpGet:path: /connectorsport: httpinitialDelaySeconds: 300periodSeconds: 10

3.3 配置 Debezium 连接器

在 Kafka Connect 中添加 Debezium 连接器，分为两部分，一个为postgres-source，从pg监控数据变化发送消息到kafka，另一个为hologres-sink，从kafka消费数据，写到hologres数据库，如下

# 添加source-connector：
curl --location --request POST 'http://kafka-connect-address:8083/connectors' \
--header 'Content-Type: application/json' \
--data-raw '{"name": "postgres-source","config": {"connector.class": "io.debezium.connector.postgresql.PostgresConnector","key.converter": "org.apache.kafka.connect.json.JsonConverter","value.converter": "org.apache.kafka.connect.json.JsonConverter","key.converter.schemas.enable": true,"value.converter.schemas.enable": true,"tasks.max": "3","database.hostname": "prod-postgresql","database.port": "5432","database.user": "postgres","database.password": "123456","database.dbname": "db_dmp","database.server.name": "prod-shuguan-dmp","schema.history.internal.kafka.bootstrap.servers": "prod-kafka:9092","schema.history.internal.kafka.topic": "pgsql-sync-hologres","table.include.list": "public.data_info,public.data_source","primary.key.fields": "id","slot.name": "debezium_slot","plugin.name": "pgoutput","publication.name": "debezium_pub","publication.autocreate.mode": "filtered","topic.prefix": "datatrans"}
}'  # 返回结果： {"name": "postgres-source","config": {"connector.class": "io.debezium.connector.postgresql.PostgresConnector","key.converter": "org.apache.kafka.connect.json.JsonConverter","value.converter": "org.apache.kafka.connect.json.JsonConverter","key.converter.schemas.enable": "true","value.converter.schemas.enable": "true","tasks.max": "3","database.hostname": "prod-postgresql","database.port": "5432","database.user": "postgres","database.password": "123456","database.dbname": "db_dmp","database.server.name": "prod-shuguan-dmp","schema.history.internal.kafka.bootstrap.servers": "prod-kafka:9092","schema.history.internal.kafka.topic": "pgsql-sync-hologres","table.include.list": "public.data_info,public.data_source","primary.key.fields": "id","slot.name": "debezium_slot","plugin.name": "pgoutput","publication.name": "debezium_pub","publication.autocreate.mode": "filtered","topic.prefix": "datatrans","name": "postgres-source"},"tasks": [],"type": "source"
} # 添加sink-connector：
curl --location --request PUT 'http://kafka-connect-address:8083/connectors/hologres-sink/config' \
--header 'Content-Type: application/json' \
--data-raw '{"name": "hologres-sink","config": {"connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector","key.converter": "org.apache.kafka.connect.json.JsonConverter","value.converter": "org.apache.kafka.connect.json.JsonConverter","key.converter.schemas.enable": true,"value.converter.schemas.enable": true,"schemas.enable": false,"tasks.max": "1","connection.url": "jdbc:postgresql://holo-cn-vpc-st.hologres.ops.auto.gee-cloud.com:80/db_dmp?reWriteBatchedInserts=true","connection.username": "admin","connection.password": "123456","insert.mode": "upsert","topics": "datatrans.public.data_info,datatrans.public.data_source","schema.history.internal.kafka.bootstrap.servers": "prod-kafka:9092","schema.history.internal.kafka.topic": "pgsql-sync-hologres","delete.enabled": "true","primary.key.fields": "id","primary.key.mode": "record_key","schema.evolution": "basic","database.time_zone": "UTC","auto.create": "true","auto.evolve": "true"}
}'# 返回结果：
{"name": "hologres-sink","config": {"connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector","key.converter": "org.apache.kafka.connect.json.JsonConverter","value.converter": "org.apache.kafka.connect.json.JsonConverter","key.converter.schemas.enable": "true","value.converter.schemas.enable": "true","schemas.enable": "false","tasks.max": "1","connection.url": "jdbc:postgresql://holo-cn-vpc-st.hologres.ops.auto.gee-cloud.com:80/db_dmp?reWriteBatchedInserts=true","connection.username": "admin","connection.password": "123456","insert.mode": "upsert","topics": "datatrans.public.data_info,datatrans.public.data_source","schema.history.internal.kafka.bootstrap.servers": "prod-kafka:9092","schema.history.internal.kafka.topic": "pgsql-sync-hologres","delete.enabled": "true","primary.key.fields": "id","primary.key.mode": "record_key","schema.evolution": "basic","database.time_zone": "UTC","auto.create": "false","auto.evolve": "false","name": "hologres-sink"},"tasks": [{"connector": "hologres-sink","task": 0}],"type": "sink"
}

删除连接器

curl --location --request DELETE 'http://kafka-connect-address:8083/connectors/postgres-source' 

查看连接器状态：

curl --location --request GET 'http://kafka-connect-address:8083/connectors/postgres-source/status' #返回结果：
{"name": "postgres-source","connector": {"state": "RUNNING","worker_id": "10.244.0.116:8083"},"tasks": [{"id": 0,"state": "RUNNING","worker_id": "10.244.0.116:8083"}],"type": "source"
}

查看连接器配置

curl --location --request GET 'http://kafka-connect-address:8083/connectors/postgres-source/config' \#返回结果：
{"connector.class": "io.debezium.connector.postgresql.PostgresConnector","publication.autocreate.mode": "filtered","database.user": "postgres","database.dbname": "db_dmp","slot.name": "debezium_slot","tasks.max": "3","publication.name": "debezium_pub","database.server.name": "prod-shuguan-dmp","schema.history.internal.kafka.bootstrap.servers": "prod-kafka:9092","database.port": "5432","plugin.name": "pgoutput","topic.prefix": "datatrans","schema.history.internal.kafka.topic": "pgsql-sync-hologres","database.hostname": "prod-postgresql","database.password": "123456","name": "postgres-source","table.include.list": "public.data_info,public.data_source"
}

查看连接器列表

curl --location --request GET 'http://kafka-connect-address:8083/connectors' #返回结果：
["postgres-source","hologres-sink"
]

其他相关接口

kafka-connect支持接口
GET /connectors #返回活动连接器的列表
POST /connectors #创建一个新的连接器; 请求主体应该是包含字符串name字段和config带有连接器配置参数的对象字段的JSON对象
GET /connectors/{name} #获取有关特定连接器的信息
GET /connectors/{name}/config #获取特定连接器的配置参数
PUT /connectors/{name}/config #更新特定连接器的配置参数
GET /connectors/{name}/status #获取连接器的当前状态，包括连接器是否正在运行，失败，已暂停等，分配给哪个工作者，失败时的错误信息以及所有任务的状态
GET /connectors/{name}/tasks #获取当前为连接器运行的任务列表
GET /connectors/{name}/tasks/{taskid}/status #获取任务的当前状态，包括如果正在运行，失败，暂停等，分配给哪个工作人员，如果失败，则返回错误信息
PUT /connectors/{name}/pause #暂停连接器及其任务，停止消息处理，直到连接器恢复
PUT /connectors/{name}/resume #恢复暂停的连接器（或者，如果连接器未暂停，则不执行任何操作）
POST /connectors/{name}/restart #重新启动连接器（通常是因为失败）
POST /connectors/{name}/tasks/{taskId}/restart #重启个别任务（通常是因为失败）
DELETE /connectors/{name} #删除连接器，停止所有任务并删除其配置#Kafka Connect还提供了用于获取有关连接器插件信息的REST API：
GET /connector-plugins #返回安装在Kafka Connect集群中的连接器插件列表。请注意，API仅检查处理请求的worker的连接器，这意味着您可能会看到不一致的结果，尤其是在滚动升级期间，如果添加新的连接器jar
PUT /connector-plugins/{connector-type}/config/validate # 根据配置定义验证提供的配置值。此API执行每个配置验证，在验证期间返回建议值和错误消息。

4. 测试和验证

4.1 数据新增

在 PostgreSQL 中插入数据：

INSERT INTO public.data_info(platform, data_type, data_name, dir_name, bag_exists, clip_exists, create_time, latest_update_time, deleted)VALUES('11v-xl', 'camera', 'aroundBack', 'cam_around_back', 1, 1, now(), now(), 0);

查看kafka-topic消息

I have no name!@prod-kafka-0:/$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic datatrans.public.data_info
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":null,"after":{"id":18,"platform":"11v-xl","data_type":"camera","data_name":"aroundBack","dir_name":"cam_around_back","bag_exists":1,"clip_exists":1,"create_time":1741311929781870,"latest_update_time":1741311929781870,"deleted":0},"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741311929783,"snapshot":"false","db":"db_dmp","sequence":"[\"12057353312\",\"12057353472\"]","ts_us":1741311929783308,"ts_ns":1741311929783308000,"schema":"public","table":"data_info","txId":3433486,"lsn":12057353472,"xmin":null},"transaction":null,"op":"c","ts_ms":1741311929873,"ts_us":1741311929873166,"ts_ns":1741311929873166000}}  

检查 Hologres 中的数据是否同步

SELECT * FROM public.data_info ORDER BY id DESC;

4.2 数据更新
在 PostgreSQL 中更新数据：

UPDATE public.data_info SET platform = '6v-xl', data_name = 'aroundBack_xl' WHERE platform = '11v-xl';

查看Kafka-topic消息：

I have no name!@prod-kafka-0:/$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic datatrans.public.data_info
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":null,"after":{"id":18,"platform":"6v-xl","data_type":"camera","data_name":"aroundBack_xl","dir_name":"cam_around_back","bag_exists":1,"clip_exists":1,"create_time":1741311929781870,"latest_update_time":1741311929781870,"deleted":0},"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741312464134,"snapshot":"false","db":"db_dmp","sequence":"[\"12065166032\",\"12065166088\"]","ts_us":1741312464134145,"ts_ns":1741312464134145000,"schema":"public","table":"data_info","txId":3434153,"lsn":12065166088,"xmin":null},"transaction":null,"op":"u","ts_ms":1741312464229,"ts_us":1741312464229054,"ts_ns":1741312464229054000}}

检查Hologres数据：

SELECT * FROM public.data_info ORDER BY id DESC;

4.3 数据删除
在 PostgreSQL 中更新数据：

DELETE FROM public.data_info WHERE platform = '6v-xl';

查看Kafka-topic消息：

I have no name!@prod-kafka-0:/$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic datatrans.public.data_info
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":{"id":18,"platform":"","data_type":"","data_name":"","dir_name":null,"bag_exists":null,"clip_exists":null,"create_time":null,"latest_update_time":null,"deleted":0},"after":null,"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741312717303,"snapshot":"false","db":"db_dmp","sequence":"[\"12068950904\",\"12068950960\"]","ts_us":1741312717303137,"ts_ns":1741312717303137000,"schema":"public","table":"data_info","txId":3434449,"lsn":12068950960,"xmin":null},"transaction":null,"op":"d","ts_ms":1741312717748,"ts_us":1741312717748938,"ts_ns":1741312717748938000}}
null

检查Hologres数据：

SELECT * FROM public.data_info ORDER BY id DESC;

4.3 批量插入
在 PostgreSQL 中插入数据：

INSERT INTO public.data_info(platform, data_type, data_name, dir_name, bag_exists, clip_exists, create_time, latest_update_time, deleted)VALUES ('11v-xl', 'camera0', 'aroundBack', 'cam_around_back', 1, 1, now(), now(), 0),('11v-xl', 'camera1', 'aroundBack', 'cam_around_back', 1, 1, now(), now(), 0), ('11v-xl', 'camera2', 'aroundBack', 'cam_around_back', 1, 1, now(), now(), 0), ('11v-xl', 'camera3', 'aroundBack', 'cam_around_back', 1, 1, now(), now(), 0), ('11v-xl', 'camera5', 'aroundBack', 'cam_around_back', 1, 1, now(), now(), 0);

查看Kafka-topic消息：

I have no name!@prod-kafka-0:/$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic datatrans.public.data_info
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":null,"after":{"id":26,"platform":"11v-xl","data_type":"camera0","data_name":"aroundBack","dir_name":"cam_around_back","bag_exists":1,"clip_exists":1,"create_time":1741313539693111,"latest_update_time":1741313539693111,"deleted":0},"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741313539694,"snapshot":"false","db":"db_dmp","sequence":"[\"12072782920\",\"12072782920\"]","ts_us":1741313539694079,"ts_ns":1741313539694079000,"schema":"public","table":"data_info","txId":3435484,"lsn":12072782920,"xmin":null},"transaction":null,"op":"c","ts_ms":1741313540003,"ts_us":1741313540003647,"ts_ns":1741313540003647000}}
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":null,"after":{"id":27,"platform":"11v-xl","data_type":"camera1","data_name":"aroundBack","dir_name":"cam_around_back","bag_exists":1,"clip_exists":1,"create_time":1741313539693111,"latest_update_time":1741313539693111,"deleted":0},"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741313539694,"snapshot":"false","db":"db_dmp","sequence":"[\"12072782920\",\"12072783208\"]","ts_us":1741313539694079,"ts_ns":1741313539694079000,"schema":"public","table":"data_info","txId":3435484,"lsn":12072783208,"xmin":null},"transaction":null,"op":"c","ts_ms":1741313540005,"ts_us":1741313540005062,"ts_ns":1741313540005062000}}
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":null,"after":{"id":29,"platform":"11v-xl","data_type":"camera3","data_name":"aroundBack","dir_name":"cam_around_back","bag_exists":1,"clip_exists":1,"create_time":1741313539693111,"latest_update_time":1741313539693111,"deleted":0},"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741313539694,"snapshot":"false","db":"db_dmp","sequence":"[\"12072782920\",\"12072783784\"]","ts_us":1741313539694079,"ts_ns":1741313539694079000,"schema":"public","table":"data_info","txId":3435484,"lsn":12072783784,"xmin":null},"transaction":null,"op":"c","ts_ms":1741313540006,"ts_us":1741313540006726,"ts_ns":1741313540006726000}}
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":null,"after":{"id":28,"platform":"11v-xl","data_type":"camera2","data_name":"aroundBack","dir_name":"cam_around_back","bag_exists":1,"clip_exists":1,"create_time":1741313539693111,"latest_update_time":1741313539693111,"deleted":0},"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741313539694,"snapshot":"false","db":"db_dmp","sequence":"[\"12072782920\",\"12072783496\"]","ts_us":1741313539694079,"ts_ns":1741313539694079000,"schema":"public","table":"data_info","txId":3435484,"lsn":12072783496,"xmin":null},"transaction":null,"op":"c","ts_ms":1741313540005,"ts_us":1741313540005913,"ts_ns":1741313540005913000}}
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":null,"after":{"id":30,"platform":"11v-xl","data_type":"camera5","data_name":"aroundBack","dir_name":"cam_around_back","bag_exists":1,"clip_exists":1,"create_time":1741313539693111,"latest_update_time":1741313539693111,"deleted":0},"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741313539694,"snapshot":"false","db":"db_dmp","sequence":"[\"12072782920\",\"12072784072\"]","ts_us":1741313539694079,"ts_ns":1741313539694079000,"schema":"public","table":"data_info","txId":3435484,"lsn":12072784072,"xmin":null},"transaction":null,"op":"c","ts_ms":1741313540007,"ts_us":1741313540007475,"ts_ns":1741313540007475000}}

检查Hologres数据：

SELECT * FROM public.data_info ORDER BY id DESC;

4.3 批量更新
在 PostgreSQL 中更新数据：

UPDATE public.data_info SET  platform = '6v-xl', data_name = 'aroundBack_xl' WHERE platform = '11v-xl';

查看Kafka-topic消息：

I have no name!@prod-kafka-0:/$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic datatrans.public.data_info
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":null,"after":{"id":26,"platform":"6v-xl","data_type":"camera0","data_name":"aroundBack_xl","dir_name":"cam_around_back","bag_exists":1,"clip_exists":1,"create_time":1741313539693111,"latest_update_time":1741313539693111,"deleted":0},"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741313734647,"snapshot":"false","db":"db_dmp","sequence":"[\"12073766384\",\"12073766440\"]","ts_us":1741313734647557,"ts_ns":1741313734647557000,"schema":"public","table":"data_info","txId":3435710,"lsn":12073766440,"xmin":null},"transaction":null,"op":"u","ts_ms":1741313734826,"ts_us":1741313734826063,"ts_ns":1741313734826063000}}
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":null,"after":{"id":27,"platform":"6v-xl","data_type":"camera1","data_name":"aroundBack_xl","dir_name":"cam_around_back","bag_exists":1,"clip_exists":1,"create_time":1741313539693111,"latest_update_time":1741313539693111,"deleted":0},"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741313734647,"snapshot":"false","db":"db_dmp","sequence":"[\"12073766384\",\"12073775280\"]","ts_us":1741313734647557,"ts_ns":1741313734647557000,"schema":"public","table":"data_info","txId":3435710,"lsn":12073775280,"xmin":null},"transaction":null,"op":"u","ts_ms":1741313734827,"ts_us":1741313734827700,"ts_ns":1741313734827700000}}
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":null,"after":{"id":29,"platform":"6v-xl","data_type":"camera3","data_name":"aroundBack_xl","dir_name":"cam_around_back","bag_exists":1,"clip_exists":1,"create_time":1741313539693111,"latest_update_time":1741313539693111,"deleted":0},"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741313734647,"snapshot":"false","db":"db_dmp","sequence":"[\"12073766384\",\"12073775872\"]","ts_us":1741313734647557,"ts_ns":1741313734647557000,"schema":"public","table":"data_info","txId":3435710,"lsn":12073775872,"xmin":null},"transaction":null,"op":"u","ts_ms":1741313734829,"ts_us":1741313734829449,"ts_ns":1741313734829449000}}
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":null,"after":{"id":28,"platform":"6v-xl","data_type":"camera2","data_name":"aroundBack_xl","dir_name":"cam_around_back","bag_exists":1,"clip_exists":1,"create_time":1741313539693111,"latest_update_time":1741313539693111,"deleted":0},"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741313734647,"snapshot":"false","db":"db_dmp","sequence":"[\"12073766384\",\"12073775576\"]","ts_us":1741313734647557,"ts_ns":1741313734647557000,"schema":"public","table":"data_info","txId":3435710,"lsn":12073775576,"xmin":null},"transaction":null,"op":"u","ts_ms":1741313734828,"ts_us":1741313734828522,"ts_ns":1741313734828522000}}
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":null,"after":{"id":30,"platform":"6v-xl","data_type":"camera5","data_name":"aroundBack_xl","dir_name":"cam_around_back","bag_exists":1,"clip_exists":1,"create_time":1741313539693111,"latest_update_time":1741313539693111,"deleted":0},"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741313734647,"snapshot":"false","db":"db_dmp","sequence":"[\"12073766384\",\"12073776168\"]","ts_us":1741313734647557,"ts_ns":1741313734647557000,"schema":"public","table":"data_info","txId":3435710,"lsn":12073776168,"xmin":null},"transaction":null,"op":"u","ts_ms":1741313734830,"ts_us":1741313734830236,"ts_ns":1741313734830236000}}

检查Hologres数据：

SELECT * FROM public.data_info ORDER BY id DESC;

4.3 批量删除
在 PostgreSQL 中删除数据：

DELETE FROM public.data_info WHERE platform = '6v-xl';

查看Kafka-topic消息：

I have no name!@prod-kafka-0:/$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic datatrans.public.data_info
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":{"id":26,"platform":"","data_type":"","data_name":"","dir_name":null,"bag_exists":null,"clip_exists":null,"create_time":null,"latest_update_time":null,"deleted":0},"after":null,"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741313871896,"snapshot":"false","db":"db_dmp","sequence":"[\"12074168712\",\"12074168768\"]","ts_us":1741313871896885,"ts_ns":1741313871896885000,"schema":"public","table":"data_info","txId":3435860,"lsn":12074168768,"xmin":null},"transaction":null,"op":"d","ts_ms":1741313871946,"ts_us":1741313871946325,"ts_ns":1741313871946325000}}
null
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":{"id":27,"platform":"","data_type":"","data_name":"","dir_name":null,"bag_exists":null,"clip_exists":null,"create_time":null,"latest_update_time":null,"deleted":0},"after":null,"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741313871896,"snapshot":"false","db":"db_dmp","sequence":"[\"12074168712\",\"12074168840\"]","ts_us":1741313871896885,"ts_ns":1741313871896885000,"schema":"public","table":"data_info","txId":3435860,"lsn":12074168840,"xmin":null},"transaction":null,"op":"d","ts_ms":1741313871947,"ts_us":1741313871947402,"ts_ns":1741313871947402000}}
null
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":{"id":29,"platform":"","data_type":"","data_name":"","dir_name":null,"bag_exists":null,"clip_exists":null,"create_time":null,"latest_update_time":null,"deleted":0},"after":null,"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741313871896,"snapshot":"false","db":"db_dmp","sequence":"[\"12074168712\",\"12074168984\"]","ts_us":1741313871896885,"ts_ns":1741313871896885000,"schema":"public","table":"data_info","txId":3435860,"lsn":12074168984,"xmin":null},"transaction":null,"op":"d","ts_ms":1741313871948,"ts_us":1741313871948261,"ts_ns":1741313871948261000}}
null
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":{"id":28,"platform":"","data_type":"","data_name":"","dir_name":null,"bag_exists":null,"clip_exists":null,"create_time":null,"latest_update_time":null,"deleted":0},"after":null,"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741313871896,"snapshot":"false","db":"db_dmp","sequence":"[\"12074168712\",\"12074168912\"]","ts_us":1741313871896885,"ts_ns":1741313871896885000,"schema":"public","table":"data_info","txId":3435860,"lsn":12074168912,"xmin":null},"transaction":null,"op":"d","ts_ms":1741313871947,"ts_us":1741313871947868,"ts_ns":1741313871947868000}}
null
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":{"id":30,"platform":"","data_type":"","data_name":"","dir_name":null,"bag_exists":null,"clip_exists":null,"create_time":null,"latest_update_time":null,"deleted":0},"after":null,"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741313871896,"snapshot":"false","db":"db_dmp","sequence":"[\"12074168712\",\"12074169056\"]","ts_us":1741313871896885,"ts_ns":1741313871896885000,"schema":"public","table":"data_info","txId":3435860,"lsn":12074169056,"xmin":null},"transaction":null,"op":"d","ts_ms":1741313871948,"ts_us":1741313871948688,"ts_ns":1741313871948688000}}
null

检查Hologres数据：

SELECT * FROM public.data_info ORDER BY id DESC;

4.4 事务操作
Pgsql先插入数据，再做更新，然后回滚更新：

BEGIN;INSERT INTO public.data_info (platform, data_type, data_name, dir_name, bag_exists, clip_exists, create_time, latest_update_time, deleted)VALUES ('11v-xl', 'camera0', 'aroundBack', 'cam_around_back', 1, 1, now(), now(), 0),('11v-xl', 'camera1', 'aroundBack', 'cam_around_back', 1, 1, now(), now(), 0), ('11v-xl', 'camera2', 'aroundBack', 'cam_around_back', 1, 1, now(), now(), 0);SAVEPOINT my_savepoint;UPDATE public.data_info SET  platform = '6v-xl', data_name = 'aroundBack_xl' WHERE platform = '11v-xl';ROLLBACK TO my_savepoint;COMMIT;

查看Kafka-topic消息，只有一次插入操作：

I have no name!@prod-kafka-0:/$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic datatrans.public.data_info
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":null,"after":{"id":37,"platform":"11v-xl","data_type":"camera0","data_name":"aroundBack","dir_name":"cam_around_back","bag_exists":1,"clip_exists":1,"create_time":1741327869321334,"latest_update_time":1741327869321334,"deleted":0},"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741327869322,"snapshot":"false","db":"db_dmp","sequence":"[\"12141193520\",\"12141193624\"]","ts_us":1741327869322708,"ts_ns":1741327869322708000,"schema":"public","table":"data_info","txId":3453096,"lsn":12141193624,"xmin":null},"transaction":null,"op":"c","ts_ms":1741327869719,"ts_us":1741327869719110,"ts_ns":1741327869719110000}}
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":null,"after":{"id":38,"platform":"11v-xl","data_type":"camera1","data_name":"aroundBack","dir_name":"cam_around_back","bag_exists":1,"clip_exists":1,"create_time":1741327869321334,"latest_update_time":1741327869321334,"deleted":0},"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741327869322,"snapshot":"false","db":"db_dmp","sequence":"[\"12141193520\",\"12141204296\"]","ts_us":1741327869322708,"ts_ns":1741327869322708000,"schema":"public","table":"data_info","txId":3453096,"lsn":12141204296,"xmin":null},"transaction":null,"op":"c","ts_ms":1741327869720,"ts_us":1741327869720542,"ts_ns":1741327869720542000}}
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":false,"field":"platform"},{"type":"string","optional":false,"field":"data_type"},{"type":"string","optional":false,"field":"data_name"},{"type":"string","optional":true,"field":"dir_name"},{"type":"int16","optional":true,"field":"bag_exists"},{"type":"int16","optional":true,"field":"clip_exists"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"create_time"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"latest_update_time"},{"type":"int16","optional":true,"default":0,"field":"deleted"}],"optional":true,"name":"datatrans5.public.data_info.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false,incremental"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"sequence"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"name":"event.block","version":1,"field":"transaction"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"int64","optional":true,"field":"ts_us"},{"type":"int64","optional":true,"field":"ts_ns"}],"optional":false,"name":"datatrans5.public.data_info.Envelope","version":2},"payload":{"before":null,"after":{"id":39,"platform":"11v-xl","data_type":"camera2","data_name":"aroundBack","dir_name":"cam_around_back","bag_exists":1,"clip_exists":1,"create_time":1741327869321334,"latest_update_time":1741327869321334,"deleted":0},"source":{"version":"2.7.4.Final","connector":"postgresql","name":"datatrans5","ts_ms":1741327869322,"snapshot":"false","db":"db_dmp","sequence":"[\"12141193520\",\"12141204584\"]","ts_us":1741327869322708,"ts_ns":1741327869322708000,"schema":"public","table":"data_info","txId":3453096,"lsn":12141204584,"xmin":null},"transaction":null,"op":"c","ts_ms":1741327869721,"ts_us":1741327869721271,"ts_ns":1741327869721271000}}

检查Hologres数据，保存UPDATE语句之前的数据：

SELECT * FROM public.data_info ORDER BY id DESC;

5. 总结

• 通过 Debezium + Kafka 的方案，可以实现 PostgreSQL 到 Hologres 的高可靠、高性能的实时 数据同步。该方案适合数据量较大、实时性要求较高的场景，同时具备良好的扩展性和灵活性。

• 数据同步写入，通过kafka-topic匹配表名，但是数据发送kafka-topic，名称必须加前缀，导致出 库和入库的表名不一致，入库的表统一多了前缀（java开发可通过mybatis拦截器，或者mybatis-plus配置解决该问题）