概述：

由于集群管理操作（如碎片分割、碎片移动、重新平衡操作）而发生的任何事件都不会导致已发布事件的发布；这是通过在此类集群管理操作期间创建的事件的WAL条目中设置replication origin字段来实现的；CDC解码器插件（pgoutput，wal2json）被添加到标准库路径中，当对逻辑复制插槽进行新的订阅时，就会加载该路径；如上所述此解码器会忽略为碎片管理操作设置了复制原点的任何event；此解码器将事件中的任何表名从已分配表的特定碎片名称转换为其相应的分布式表名称；

有两种重要类型的CDC客户端：

PostgreSQL客户端从分布式表中复制数据像Apache Kafka这样的事件流消费者

Postgresql客户端CDC

启用CDC功能

分布式表的CDC功能正在预览（测试版）中，应该明确启用。应该在postgresql.conf中设置此标志，以便为所有后端启用它。

citus.enable_change_data_capture='on'

创建分布式表

CREATE TABLE sensors(

measureid integer,

eventdatetime timestamptz,

measure_data jsonb,

meaure_quantity decimal(15, 2),

measure_status char(1),

measure_comment varchar(44),

PRIMARY KEY (measureid, eventdatetime, measure_data)

);

SELECT create_distributed_table('sensors','measureid');

在PG客户端创建Schema相同的表

创建publication

在Coordinate节点上的分布式表上创建发布，会自动将publication同步到所有worker节点；

CREATE PUBLICATION cdc_publication FOR TABLE sensors;

创建逻辑复制槽

SELECT * FROM run_command_on_all_nodes

($$SELECT pg_create_logical_replication_slot('cdc_replication_slot', 'pgoutput', false);$$);

在CDC客户端创建subscriptions

实现方式一：创建对Coordinate节点的订阅

CREATE SUBSCRIPTION cdc_subscription_1

CONNECTION 'dbname= host= user= port='

PUBLICATION

WITH (copy_data=true,create_slot=false,slot_name='');

参数说明：

--dbname: Name of the database where the distributed table to be replicated resides.

--username: Database user name.

--coordinator host: hostname of the co-ordinator node

--coordinator port: port of the co-ordinator node.

--cdc_publication_name: publication created in the co-ordinator for the distributed table.

--cdc_replication_slot: logical replication slot in co-ordinator node for CDC.

NOTE: 'copy_data' argument should be set to true to copy any existing data to the CDC client.

This should be done for only one subscription in CDC client otherwise it will result in duplicate data in CDC client and cause replication errors.

实现方式二：创建对Worker节点的订阅

CREATE SUBSCRIPTION cdc_subscription_2

CONNECTION 'dbname= host= user= port='

PUBLICATION cdc_publication

WITH (copy_data=false,create_slot=false,slot_name='cdc_replication_slot');

CREATE SUBSCRIPTION cdc_subscription_3

CONNECTION 'dbname= host= user= port='

PUBLICATION cdc_publication

WITH (copy_data=false,create_slot=false,slot_name='cdc_replication_slot');

NOTE: 'copy_data' argument should be set to false to avoid copying existing data into CDC client from every worker node because that will result in duplicate data and cause replication errors.

验证订阅是否成功

Kafka客户端CDC

分布式表的配置参考上面的操作

创建debezium连接器配置

对于分布式表，应为每个工作节点创建一个debezium连接器配置文件，其中包含每个工作节点中分布式表的发布和复制槽信息的详细信息；

name=dbz-test-connector

connector.class=io.debezium.connector.postgresql.PostgresConnector

tasks.max=1

plugin.name=pgoutput

database.hostname=

database.port=

database.user=

database.password=

database.dbname=

database.server.name =

slot.name=

publication.name=

key.converter=org.apache.kafka.connect.json.JsonConverter

value.converter=org.apache.kafka.connect.json.JsonConverter

key.converter.schemas.enable=false

value.converter.schemas.enable=false

NOTE: For getting changes to reference table the CDC client should connnect only to the co-ordinator node,

since the changes for Reference tables are published only from co-ordinator node.

Citus 11.3版本使用限制

限制1：继承所有Postgres逻辑复制限制

Citus CDC正在使用逻辑复制，因此逻辑复制的任何限制也适用于CDC。逻辑复制的一些重要限制是：

DDL命令不会通过逻辑复制进行复制。因此，对分布式表模式的任何更改都必须手动应用于CDC客户端。

主标识或复制标识应设置为FULL，以便将对表的所有更改发布到CDC客户端。

CDC客户端的复制目标必须是常规PostgreSQL表，不能是分布式表。

限制2：无法保证跨节点的事件排序

从同一工作节点发布的事件顺序与该节点中发生的事件顺序相同。但是，对于跨多个工作节点发生的事件的顺序没有任何保证。

限制3：CDC的初始快照可能不一致

使用copy_data=true为分布式表拍摄的初始快照可能会导致事件发布无序，因为在将初始数据发布到CDC客户端时，数据可能会在工作节点之间发生更改。因此，建议在稳定状态下对CDC进行初步快照。

限制4：CDC不适用于基于列存储的表格

限制5：CDC订阅所需的每个工作节点

必须创建从CDC客户端到每个工作节点的单独订阅，以获取分布式表的所有工作节点上发生的事件

限制6：CDC不支持ALL TABLES发布

如果使用CREATE publication with ALL TABLES 来创建发布，则还会复制所有Citus内部元数据表（如pg_dist_partition）。在这种情况下，如果您要从Citus复制到Postgres，则应确保所有元数据表都是在订阅服务器端创建的。从Citus 11.3开始，我们不鼓励在CREATE PUBLICATION命令中使用ALL TABLES语法。

限制7：Reference table changes published from co-ordinator only

Reference表事件仅从协调器节点发布，以避免发布重复事件，因为引用表已复制到所有工作节点。因此，CDC客户端必须订阅coordinator节点才能获得Reference表的更改。

参考资料