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

飞不高  论坛元老 | 2025-3-20 15:45:10 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 1539|帖子 1539|积分 4617

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要 登录 才可以下载或查看，没有账号？立即注册

x
基于 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

启用逻辑复制：
  1. ALTER SYSTEM SET wal_level = logical;
复制代码
创建复制槽：
  1. # 创建复制槽（replication slot）：
  2. SELECT pg_create_logical_replication_slot('debezium_slot', 'pgoutput');
复制代码
  1. # 查看复制插槽
  2. SELECT * FROM pg_replication_slots;
复制代码
创建发布：
  1. # 创建发布(配置连接器可自动创建）
  2. CREATE PUBLICATION debezium_pub FOR ALL TABLES;
  3. CREATE PUBLICATION debezium_pub FOR TABLE my_table1, my_table2;
  5. # 删除发布
  6. DROP PUBLICATION IF EXISTS debezium_pub;
复制代码
确保 PostgreSQL 用户具有复制权限：
  1. # 最好使用superuser
  2. 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/
  1. 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/
  1. # 创建插件目录（如果不存在）
  2. sudo mkdir -p /usr/share/java/kafka-connect/
  4. # 复制 JAR 文件
  5. 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 文件中挂载插件目录：
  1. volumes:
  2.         - emptyDir: {}
  3.           name: plugins-volume
  4.         - emptyDir: {}
  5.           name: config-volume
  7. ​ volumeMounts:
  8.         - name: plugins-volume
  9.           mountPath: /usr/share/java/kafka-connect/
  10.         - name: config-volume
  11.           mountPath: /etc/kafka-connect/
复制代码
Kafka-connect对应的Service、StatefulSets对象的yaml
  1. apiVersion: v1
  2. kind: Service
  3. metadata:
  4.   name: prod-kafka-connect
  5.   namespace: prod-core-jobs-shuguan
  6. spec:
  7.   ports:
  8.     - port: 8083
  9.       targetPort: 8083
  10.       protocol: TCP
  11.       name: http
  12.   selector:
  13.     app: prod-kafka-connect
  14.   type: ClusterIP
  16. ---
  17. apiVersion: apps/v1
  18. kind: StatefulSet
  19. metadata:
  20.   name: prod-kafka-connect
  21.   namespace: prod-core-jobs-shuguan
  22. spec:
  23.   serviceName: "prod-kafka-connect"
  24.   replicas: 1
  25.   selector:
  26.     matchLabels:
  27.       app: prod-kafka-connect
  28.   template:
  29.     metadata:
  30.       labels:
  31.         app: prod-kafka-connect
  32.     spec:
  33.       volumes:
  34.         - emptyDir: {}
  35.           name: plugins-volume
  36.         - emptyDir: {}
  37.           name: config-volume
  38.       initContainers:
  39.         - name: extract-plugins
  40.           image: pkg.geely.com/docker/busybox:1.31   # 初始化容器，作下载拷贝Debezium插件使用
  41.           command:
  42.             - sh
  43.             - -c
  44.             - |
  45.               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.gz
  46.               tar -zxvf /etc/kafka-connect/debezium-connector-postgres-2.2.0.Final-plugin.tar.gz -C /usr/share/java/kafka-connect/
  47.           volumeMounts:
  48.             - name: plugins-volume
  49.               mountPath: /usr/share/java/kafka-connect/
  50.             - name: config-volume
  51.               mountPath: /etc/kafka-connect/
  52.       containers:
  53.         - name: prod-kafka-connect
  54.           image: pkg.geely.com/docker/confluentinc/cp-kafka-connect:7.2.0
  55.           ports:
  56.             - containerPort: 8083
  57.               name: http
  58.           env:
  59.             - name: CONNECT_BOOTSTRAP_SERVERS
  60.               value: "prod-kafka:9092"  # 替换为你的 Kafka 集群地址
  61.             - name: CONNECT_GROUP_ID
  62.               value: "connect-cluster"
  63.             - name: CONNECT_KEY_CONVERTER
  64.               value: "org.apache.kafka.connect.storage.StringConverter"
  65.             - name: CONNECT_VALUE_CONVERTER
  66.               value: "org.apache.kafka.connect.storage.StringConverter"
  67.             - name: CONNECT_CONFIG_STORAGE_TOPIC
  68.               value: "connect-configs"
  69.             - name: CONNECT_OFFSET_STORAGE_TOPIC
  70.               value: "connect-offsets"
  71.             - name: CONNECT_STATUS_STORAGE_TOPIC
  72.               value: "connect-statuses"
  73.             - name: CONNECT_REST_ADVERTISED_HOST_NAME
  74.               valueFrom:
  75.                 fieldRef:
  76.                   fieldPath: status.podIP
  77.             - name: CONNECT_REST_ADVERTISED_PORT
  78.               value: "8083"
  79.           volumeMounts:
  80.             - name: plugins-volume
  81.               mountPath: /usr/share/java/kafka-connect/
  82.             - name: config-volume
  83.               mountPath: /etc/kafka-connect/
  84.           readinessProbe:
  85.             httpGet:
  86.               path: /connectors
  87.               port: http
  88.             initialDelaySeconds: 30
  89.             periodSeconds: 10
  90.           livenessProbe:
  91.             httpGet:
  92.               path: /connectors
  93.               port: http
  94.             initialDelaySeconds: 300
  95.             periodSeconds: 10
复制代码
3.3 设置 Debezium 毗连器

在 Kafka Connect 中添加 Debezium 毗连器，分为两部分，一个为postgres-source，从pg监控数据变化发送消息到kafka，另一个为hologres-sink，从kafka消费数据，写到hologres数据库，如下
  1. # 添加source-connector：
  2. curl --location --request POST 'http://kafka-connect-address:8083/connectors' \
  3. --header 'Content-Type: application/json' \
  4. --data-raw '{
  5.   "name": "postgres-source",
  6.   "config": {
  7.     "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
  8.     "key.converter": "org.apache.kafka.connect.json.JsonConverter",
  9.     "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  10.     "key.converter.schemas.enable": true,
  11.     "value.converter.schemas.enable": true,
  12.     "tasks.max": "3",
  13.     "database.hostname": "prod-postgresql",
  14.     "database.port": "5432",
  15.     "database.user": "postgres",
  16.     "database.password": "123456",
  17.     "database.dbname": "db_dmp",
  18.     "database.server.name": "prod-shuguan-dmp",
  19.     "schema.history.internal.kafka.bootstrap.servers": "prod-kafka:9092",
  20.     "schema.history.internal.kafka.topic": "pgsql-sync-hologres",
  21.     "table.include.list": "public.data_info,public.data_source",
  22.     "primary.key.fields": "id",
  23.     "slot.name": "debezium_slot",
  24.     "plugin.name": "pgoutput",
  25.     "publication.name": "debezium_pub",
  26.     "publication.autocreate.mode": "filtered",
  27.     "topic.prefix": "datatrans"
  28.   }
  29. }'  
  31. # 返回结果： {
  32.     "name": "postgres-source",
  33.     "config": {
  34.         "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
  35.         "key.converter": "org.apache.kafka.connect.json.JsonConverter",
  36.         "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  37.         "key.converter.schemas.enable": "true",
  38.         "value.converter.schemas.enable": "true",
  39.         "tasks.max": "3",
  40.         "database.hostname": "prod-postgresql",
  41.         "database.port": "5432",
  42.         "database.user": "postgres",
  43.         "database.password": "123456",
  44.         "database.dbname": "db_dmp",
  45.         "database.server.name": "prod-shuguan-dmp",
  46.         "schema.history.internal.kafka.bootstrap.servers": "prod-kafka:9092",
  47.         "schema.history.internal.kafka.topic": "pgsql-sync-hologres",
  48.         "table.include.list": "public.data_info,public.data_source",
  49.         "primary.key.fields": "id",
  50.         "slot.name": "debezium_slot",
  51.         "plugin.name": "pgoutput",
  52.         "publication.name": "debezium_pub",
  53.         "publication.autocreate.mode": "filtered",
  54.         "topic.prefix": "datatrans",
  55.         "name": "postgres-source"
  56.     },
  57.     "tasks": [],
  58.     "type": "source"
  59. }
  61. # 添加sink-connector：
  62. curl --location --request PUT 'http://kafka-connect-address:8083/connectors/hologres-sink/config' \
  63. --header 'Content-Type: application/json' \
  64. --data-raw '{
  65.     "name": "hologres-sink",
  66.     "config": {
  67.         "connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector",
  68.         "key.converter": "org.apache.kafka.connect.json.JsonConverter",
  69.         "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  70.         "key.converter.schemas.enable": true,
  71.         "value.converter.schemas.enable": true,
  72.         "schemas.enable": false,
  73.         "tasks.max": "1",
  74.         "connection.url": "jdbc:postgresql://holo-cn-vpc-st.hologres.ops.auto.gee-cloud.com:80/db_dmp?reWriteBatchedInserts=true",
  75.         "connection.username": "admin",
  76.         "connection.password": "123456",
  77.         "insert.mode": "upsert",
  78.         "topics": "datatrans.public.data_info,datatrans.public.data_source",
  79.         "schema.history.internal.kafka.bootstrap.servers": "prod-kafka:9092",
  80.         "schema.history.internal.kafka.topic": "pgsql-sync-hologres",
  81.         "delete.enabled": "true",
  82.         "primary.key.fields": "id",
  83.         "primary.key.mode": "record_key",
  84.         "schema.evolution": "basic",
  85.         "database.time_zone": "UTC",
  86.         "auto.create": "true",
  87.         "auto.evolve": "true"
  88.     }
  89. }'
  91. # 返回结果：
  92. {
  93.     "name": "hologres-sink",
  94.     "config": {
  95.         "connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector",
  96.         "key.converter": "org.apache.kafka.connect.json.JsonConverter",
  97.         "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  98.         "key.converter.schemas.enable": "true",
  99.         "value.converter.schemas.enable": "true",
  100.         "schemas.enable": "false",
  101.         "tasks.max": "1",
  102.         "connection.url": "jdbc:postgresql://holo-cn-vpc-st.hologres.ops.auto.gee-cloud.com:80/db_dmp?reWriteBatchedInserts=true",
  103.         "connection.username": "admin",
  104.         "connection.password": "123456",
  105.         "insert.mode": "upsert",
  106.         "topics": "datatrans.public.data_info,datatrans.public.data_source",
  107.         "schema.history.internal.kafka.bootstrap.servers": "prod-kafka:9092",
  108.         "schema.history.internal.kafka.topic": "pgsql-sync-hologres",
  109.         "delete.enabled": "true",
  110.         "primary.key.fields": "id",
  111.         "primary.key.mode": "record_key",
  112.         "schema.evolution": "basic",
  113.         "database.time_zone": "UTC",
  114.         "auto.create": "false",
  115.         "auto.evolve": "false",
  116.         "name": "hologres-sink"
  117.     },
  118.     "tasks": [
  119.         {
  120.             "connector": "hologres-sink",
  121.             "task": 0
  122.         }
  123.     ],
  124.     "type": "sink"
  125. }
复制代码
删除毗连器
  1. curl --location --request DELETE 'http://kafka-connect-address:8083/connectors/postgres-source'
复制代码
查察毗连器状态：
  1. curl --location --request GET 'http://kafka-connect-address:8083/connectors/postgres-source/status'
  3. #返回结果：
  4. {
  5.     "name": "postgres-source",
  6.     "connector": {
  7.         "state": "RUNNING",
  8.         "worker_id": "10.244.0.116:8083"
  9.     },
  10.     "tasks": [
  11.         {
  12.             "id": 0,
  13.             "state": "RUNNING",
  14.             "worker_id": "10.244.0.116:8083"
  15.         }
  16.     ],
  17.     "type": "source"
  18. }
复制代码
查察毗连器设置
  1. curl --location --request GET 'http://kafka-connect-address:8083/connectors/postgres-source/config' \
  3. #返回结果：
  4. {
  5.     "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
  6.     "publication.autocreate.mode": "filtered",
  7.     "database.user": "postgres",
  8.     "database.dbname": "db_dmp",
  9.     "slot.name": "debezium_slot",
  10.     "tasks.max": "3",
  11.     "publication.name": "debezium_pub",
  12.     "database.server.name": "prod-shuguan-dmp",
  13.     "schema.history.internal.kafka.bootstrap.servers": "prod-kafka:9092",
  14.     "database.port": "5432",
  15.     "plugin.name": "pgoutput",
  16.     "topic.prefix": "datatrans",
  17.     "schema.history.internal.kafka.topic": "pgsql-sync-hologres",
  18.     "database.hostname": "prod-postgresql",
  19.     "database.password": "123456",
  20.     "name": "postgres-source",
  21.     "table.include.list": "public.data_info,public.data_source"
  22. }
复制代码
查察毗连器列表
  1. curl --location --request GET 'http://kafka-connect-address:8083/connectors'
  4. #返回结果：
  5. [
  6.     "postgres-source",
  7.     "hologres-sink"
  8. ]
复制代码
其他相关接口
  1. kafka-connect支持接口
  2. GET /connectors #返回活动连接器的列表
  3. POST /connectors #创建一个新的连接器; 请求主体应该是包含字符串name字段和config带有连接器配置参数的对象字段的JSON对象
  4. GET /connectors/{name} #获取有关特定连接器的信息
  5. GET /connectors/{name}/config #获取特定连接器的配置参数
  6. PUT /connectors/{name}/config #更新特定连接器的配置参数
  7. GET /connectors/{name}/status #获取连接器的当前状态，包括连接器是否正在运行，失败，已暂停等，分配给哪个工作者，失败时的错误信息以及所有任务的状态
  8. GET /connectors/{name}/tasks #获取当前为连接器运行的任务列表
  9. GET /connectors/{name}/tasks/{taskid}/status #获取任务的当前状态，包括如果正在运行，失败，暂停等，分配给哪个工作人员，如果失败，则返回错误信息
  10. PUT /connectors/{name}/pause #暂停连接器及其任务，停止消息处理，直到连接器恢复
  11. PUT /connectors/{name}/resume #恢复暂停的连接器（或者，如果连接器未暂停，则不执行任何操作）
  12. POST /connectors/{name}/restart #重新启动连接器（通常是因为失败）
  13. POST /connectors/{name}/tasks/{taskId}/restart #重启个别任务（通常是因为失败）
  14. DELETE /connectors/{name} #删除连接器，停止所有任务并删除其配置
  16. #Kafka Connect还提供了用于获取有关连接器插件信息的REST API：
  17. GET /connector-plugins #返回安装在Kafka Connect集群中的连接器插件列表。请注意，API仅检查处理请求的worker的连接器，这意味着您可能会看到不一致的结果，尤其是在滚动升级期间，如果添加新的连接器jar
  18. PUT /connector-plugins/{connector-type}/config/validate # 根据配置定义验证提供的配置值。此API执行每个配置验证，在验证期间返回建议值和错误消息。
复制代码
4. 测试和验证

4.1 数据新增

在 PostgreSQL 中插入数据：
  1. INSERT INTO public.data_info(
  2.         platform, data_type, data_name, dir_name, bag_exists, clip_exists, create_time, latest_update_time, deleted)
  3.         VALUES('11v-xl', 'camera', 'aroundBack', 'cam_around_back', 1, 1, now(), now(), 0);
复制代码
查察kafka-topic消息
  1. I have no name!@prod-kafka-0:/$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic datatrans.public.data_info
  2. {"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 中的数据是否同步
  1. SELECT * FROM public.data_info ORDER BY id DESC;
复制代码
4.2 数据更新
在 PostgreSQL 中更新数据：
  1. UPDATE public.data_info SET platform = '6v-xl', data_name = 'aroundBack_xl' WHERE platform = '11v-xl';
复制代码
查察Kafka-topic消息：
  2. I have no name!@prod-kafka-0:/$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic datatrans.public.data_info
  3. {"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数据：
  1. SELECT * FROM public.data_info ORDER BY id DESC;
复制代码
4.3 数据删除
在 PostgreSQL 中更新数据：
  1. DELETE FROM public.data_info WHERE platform = '6v-xl';
复制代码
查察Kafka-topic消息：
  1. I have no name!@prod-kafka-0:/$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic datatrans.public.data_info
  2. {"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}}
  3. null
复制代码
检查Hologres数据：
  1. SELECT * FROM public.data_info ORDER BY id DESC;
复制代码
4.3 批量插入
在 PostgreSQL 中插入数据：
  1. INSERT INTO public.data_info(
  2.         platform, data_type, data_name, dir_name, bag_exists, clip_exists, create_time, latest_update_time, deleted)
  3.         VALUES ('11v-xl', 'camera0', 'aroundBack', 'cam_around_back', 1, 1, now(), now(), 0),
  4.         ('11v-xl', 'camera1', 'aroundBack', 'cam_around_back', 1, 1, now(), now(), 0),
  5.         ('11v-xl', 'camera2', 'aroundBack', 'cam_around_back', 1, 1, now(), now(), 0),
  6.         ('11v-xl', 'camera3', 'aroundBack', 'cam_around_back', 1, 1, now(), now(), 0),
  7.         ('11v-xl', 'camera5', 'aroundBack', 'cam_around_back', 1, 1, now(), now(), 0);
复制代码
查察Kafka-topic消息：
  1. I have no name!@prod-kafka-0:/$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic datatrans.public.data_info
  2. {"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}}
  3. {"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}}
  4. {"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}}
  5. {"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}}
  6. {"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数据：
  1. SELECT * FROM public.data_info ORDER BY id DESC;
复制代码
4.3 批量更新
在 PostgreSQL 中更新数据：
  1. UPDATE public.data_info SET  platform = '6v-xl', data_name = 'aroundBack_xl' WHERE platform = '11v-xl';
复制代码
查察Kafka-topic消息：
  1. I have no name!@prod-kafka-0:/$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic datatrans.public.data_info
  2. {"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}}
  3. {"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}}
  4. {"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}}
  5. {"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}}
  6. {"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数据：
  1. SELECT * FROM public.data_info ORDER BY id DESC;
复制代码
4.3 批量删除
在 PostgreSQL 中删除数据：
  1. DELETE FROM public.data_info WHERE platform = '6v-xl';
复制代码
查察Kafka-topic消息：
  1. I have no name!@prod-kafka-0:/$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic datatrans.public.data_info
  2. {"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}}
  3. null
  4. {"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}}
  5. null
  6. {"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}}
  7. null
  8. {"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}}
  9. null
  10. {"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}}
  11. null
复制代码
检查Hologres数据：
  1. SELECT * FROM public.data_info ORDER BY id DESC;
复制代码
4.4 事务操作
Pgsql先插入数据，再做更新，然后回滚更新：
  1. 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';
  2. ROLLBACK TO my_savepoint;COMMIT;
复制代码
查察Kafka-topic消息，只有一次插入操作：
  1. I have no name!@prod-kafka-0:/$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic datatrans.public.data_info
  2. {"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}}
  3. {"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}}
  4. {"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语句之前的数据：
  1. SELECT * FROM public.data_info ORDER BY id DESC;
复制代码
5. 总结



  • 通过 Debezium + Kafka 的方案，可以实现 PostgreSQL 到 Hologres 的高可靠、高性能的实时 数据同步。该方案适合数据量较大、实时性要求较高的场景，同时具备精良的扩展性和机动性。
  • 数据同步写入，通过kafka-topic匹配表名，但是数据发送kafka-topic，名称必须加前缀，导致出 库和入库的表名不一致，入库的表统一多了前缀（java开辟可通过mybatis拦截器，或者mybatis-plus设置办理该问题）

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

飞不高

论坛元老
这个人很懒什么都没写!

楼主热帖

标签云

AI 运维 CIO 存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表