- 动态与公告
- 产品简介
- 购买指南
- 快速入门
- 准备工作
- 数据迁移
- 数据同步
- 数据订阅（Kafka 版）
- 前置校验不通过处理方法
- 割接说明
- 监控与告警
- 运维管理
- 实践教程
- 常见问题
- 错误处理
- API 文档
- History
- Introduction
- Introduction
- API Category
- Making API Requests
- (NewDTS) Data Migration APIs
- DescribeMigrationJobs
- CreateMigrationService
- ModifyMigrationJob
- CreateMigrateCheckJob
- DescribeMigrationCheckJob
- SkipCheckItem
- StartMigrateJob
- CompleteMigrateJob
- ContinueMigrateJob
- DescribeMigrateDBInstances
- DescribeMigrationDetail
- DestroyMigrateJob
- IsolateMigrateJob
- ModifyMigrateJobSpec
- ModifyMigrateName
- ModifyMigrateRuntimeAttribute
- PauseMigrateJob
- RecoverMigrateJob
- ResumeMigrateJob
- StopMigrateJob
- ModifyMigrateRateLimit
- Data Sync APIs
- DescribeSyncJobs
- CreateSyncJob
- ConfigureSyncJob
- CreateCheckSyncJob
- DescribeCheckSyncJobResult
- SkipSyncCheckItem
- StartSyncJob
- IsolateSyncJob
- RecoverSyncJob
- PauseSyncJob
- ContinueSyncJob
- ResumeSyncJob
- StopSyncJob
- DestroySyncJob
- ResizeSyncJob
- ModifySyncJobConfig
- CreateModifyCheckSyncJob
- DescribeModifyCheckSyncJobResult
- StartModifySyncJob
- ModifySyncRateLimit
- Data Consistency Check APIs
- (NewDTS) Data Subscription APIs
- CreateSubscribe
- ConfigureSubscribeJob
- CreateSubscribeCheckJob
- DescribeSubscribeCheckJob
- StartSubscribe
- ModifySubscribeObjects
- DescribeSubscribeDetail
- DescribeSubscribeJobs
- ResumeSubscribe
- CreateConsumerGroup
- DescribeConsumerGroups
- ModifyConsumerGroupDescription
- ModifyConsumerGroupPassword
- DescribeOffsetByTime
- ResetConsumerGroupOffset
- DeleteConsumerGroup
- ResetSubscribe
- ModifySubscribeAutoRenewFlag
- ModifySubscribeName
- DescribeSubscribeReturnable
- IsolateSubscribe
- DestroyIsolatedSubscribe
- Data Types
- Error Codes
- DTS API 2018-03-30
- 相关协议
- 动态与公告
- 产品简介
- 购买指南
- 快速入门
- 准备工作
- 数据迁移
- 数据同步
- 数据订阅（Kafka 版）
- 前置校验不通过处理方法
- 割接说明
- 监控与告警
- 运维管理
- 实践教程
- 常见问题
- 错误处理
- API 文档
- History
- Introduction
- Introduction
- API Category
- Making API Requests
- (NewDTS) Data Migration APIs
- DescribeMigrationJobs
- CreateMigrationService
- ModifyMigrationJob
- CreateMigrateCheckJob
- DescribeMigrationCheckJob
- SkipCheckItem
- StartMigrateJob
- CompleteMigrateJob
- ContinueMigrateJob
- DescribeMigrateDBInstances
- DescribeMigrationDetail
- DestroyMigrateJob
- IsolateMigrateJob
- ModifyMigrateJobSpec
- ModifyMigrateName
- ModifyMigrateRuntimeAttribute
- PauseMigrateJob
- RecoverMigrateJob
- ResumeMigrateJob
- StopMigrateJob
- ModifyMigrateRateLimit
- Data Sync APIs
- DescribeSyncJobs
- CreateSyncJob
- ConfigureSyncJob
- CreateCheckSyncJob
- DescribeCheckSyncJobResult
- SkipSyncCheckItem
- StartSyncJob
- IsolateSyncJob
- RecoverSyncJob
- PauseSyncJob
- ContinueSyncJob
- ResumeSyncJob
- StopSyncJob
- DestroySyncJob
- ResizeSyncJob
- ModifySyncJobConfig
- CreateModifyCheckSyncJob
- DescribeModifyCheckSyncJobResult
- StartModifySyncJob
- ModifySyncRateLimit
- Data Consistency Check APIs
- (NewDTS) Data Subscription APIs
- CreateSubscribe
- ConfigureSubscribeJob
- CreateSubscribeCheckJob
- DescribeSubscribeCheckJob
- StartSubscribe
- ModifySubscribeObjects
- DescribeSubscribeDetail
- DescribeSubscribeJobs
- ResumeSubscribe
- CreateConsumerGroup
- DescribeConsumerGroups
- ModifyConsumerGroupDescription
- ModifyConsumerGroupPassword
- DescribeOffsetByTime
- ResetConsumerGroupOffset
- DeleteConsumerGroup
- ResetSubscribe
- ModifySubscribeAutoRenewFlag
- ModifySubscribeName
- DescribeSubscribeReturnable
- IsolateSubscribe
- DestroyIsolatedSubscribe
- Data Types
- Error Codes
- DTS API 2018-03-30
- 相关协议
本场景介绍使用 DTS 创建腾讯云 TDSQL PostgreSQL 版的数据订阅任务操作指导。
前提条件
已准备好待订阅的 TDSQL PostgreSQL 版，并且数据库版本符合要求，请参见 数据订阅支持的数据库。
已在源端实例中创建订阅账号，需要账号权限如下： LOGIN 和 REPLICATION 权限。
LOGIN 和 REPLICATION 授权请 提交工单 处理。
订阅账号必须拥有被订阅表的 select 权限，如果是整库订阅，那么订阅账号要拥有该 schema 下所有表的 select 权限，具体授权语句如下：
grant SELECT on all tables in schema "schema_name" to '迁移账号' ;
用户必须拥有 pg_catalog.pgxc_node 表的 select 权限，具体授权语句如下：
grant SELECT on pg_catalog.pgxc_node to '迁移账号';
DN 节点的 wal_level 必须是 logical。
被订阅的表如果是全复制表（建表语句中有 distribute by replication 关键字），必须拥有主键；被订阅的表如果不是全复制表，必须拥有主键或 REPLICA IDENTITY 为 FULL；修改表的 REPLICA IDENTITY 为 FULL 的语句：
alter table '表名' REPLICA IDENTITY FULL;
约束限制
订阅的消息保存在 DTS 内置 Kafka（单 Topic）中，目前默认保存时间为最近1天，单 Topic 的最大存储为500G，当数据存储时间超过1天，或者数据量超过500G时，内置 Kafka 都会开始清除最先写入的数据。所以请用户及时消费，避免数据在消费完之前就被清除。
数据消费的地域需要与订阅任务所属的地域相同。
当前不支持 gtsvector, pg_dependencies, pg_node_tree, pg_ndistinct, xml 相关的数据类型。
数据订阅源是 TDSQL PostgreSQL 版时，不支持直接执行授权语句授权，所以订阅账号的权限需要在 TDSQL 控制台 单击实例 ID，获取实例登录信息后，通过客户端登录数据库进行账号授权。
订阅任务过程中，如果进行修改订阅对象等操作会发生任务重启，重启后可能会导致用户在 kafka 客户端消费数据时出现重复。
DTS 是按最小数据单元进行传输的，增量数据每标记一个 checkpoint 位点就是一个数据单元，如果重启时，刚好一个数据单元传输已完成，则不会导致数据重复；如果重启时，一个数据单元还正在传输中，那么再次启动后需要重新拉取这个数据单元，以保证数据完整性，这样就会导致数据重复。
用户如果对重复数据比较关注，请自行在消费数据时设置去重逻辑。
支持订阅的 SQL 操作
操作类型 | 支持的 SQL 操作 |
DML | INSERT、UPDATE、DELETE |
操作步骤
1. 登录 DTS 控制台，在左侧导航选择数据订阅，单击新建数据订阅。
2. 在新建数据订阅页，选择相应配置，单击立即购买。
计费模式：支持包年包月和按量计费。
地域：地域需与待订阅的数据库实例订阅保持一致。
数据库：请根据具体数据库类型进行选择。
版本：选择 kafka 版，支持通过 Kafka 客户端直接消费。
订阅实例名称：编辑当前数据订阅实例的名称。
3. 购买成功后，返回数据订阅列表，单击操作列的配置订阅对刚购买的订阅进行配置，配置完成后才可以进行使用。
4. 在配置数据订阅页面，选择相应配置，单击下一步。
实例：选择对应数据库实例，只读实例和灾备实例不支持数据订阅。
数据库账号：添加订阅实例的账号和密码，账号的 LOGIN 、REPLICATION 权限和全部对象、pg_catalog.pgxc_node 表的 SELECT 权限。
5. 在订阅类型和对象选择页面，选择订阅类型，单击保存配置。
订阅类型为数据更新（订阅选择对象的数据更新，包括数据 INSERT、UPDATE、DELETE 操作）。
Kafka 分区策略：支持按表名分区。
6. 在预校验页面，预校验任务预计会运行2分钟 - 3分钟，预校验通过后，单击启动完成数据订阅任务配置。
说明：
如果校验失败，请根据失败提示在待订阅实例中进行修正，并重新进行校验。
7. 单击启动后，订阅任务会进行初始化，预计会运行3分钟 - 4分钟，初始化成功后进入运行中状态。
8. 新增消费组，数据订阅 Kafka 版支持用户创建多个消费组，进行多点消费。数据订阅 Kafka 版消费依赖于 Kafka 的消费组，所以在消费数据前需要创建消费组。
9. 订阅实例进入运行中状态之后，就可以开始消费数据。Kafka 的消费需要进行密码认证，具体示例请参考 数据消费 Demo，我们提供了多种语言的 Demo 代码，也对消费的主要流程和关键的数据结构进行了说明。
是
否
本页内容是否解决了您的问题？