简介阿里云专有云Enterprise版V3.16.0的数据传输服务DTS开发指南面向企业开发者、运维人员及架构师帮助在专有云环境中通过API接口完成数据迁移、同步与订阅任务的开发集成。文档版本日期为20220301正文依次说明法律声明、通用约定、快速入门和准备工作覆盖登录API与工具控制台、获取AccessKey与STS AccessKey、公共Header参数以及DTS Endpoint等前置步骤。Java、Python、Go三种语言的SDK调用示例均有专门章节包括安装SDK、设置身份验证凭证、请求连接配置、发起调用与错误处理适合不同技术栈的研发团队对照使用。新版API参考是核心内容详细给出创建DTS实例、配置迁移或同步任务、配置订阅任务、消费组管理、批量启动、查询任务详情与修改任务等接口的功能、参数、请求方法和响应参数。整个资源共1个PDF文件压缩包3.57MB结构清晰目前已有70人学习可作为专有云DTS二次开发的接口速查与实施参考。1. 专有云里的 DTS 开发指南先搞清楚它和公共云差在哪做阿里云数据迁移和同步的工程师都有个共识公共云上的 DTS 控制台点几下就能用但到了专有云 Enterprise版Apsara Stack里事情立刻变复杂。你面对的不再是「开通服务 → 创建任务 → 看监控」的三步走而是一套部署在企业内网里的完整数据传输服务需要自己申请资源、配置网络、管理任务 API甚至要处理平台版本升级带来的兼容问题。这份 V3.16.0 开发指南本质上就是给在专有云环境里做数据接入、迁移、同步的开发者和运维人员看的接口手册与落地规范。你可能会问专有云和公共云的 DTS 差别到底有多大这么说吧公共云你调的是阿里云开放 API走公网 endpoint用 RAM 账号做权限控制专有云里 endpoint 是内网地址认证方式可能走 token 或内部 AK/SK任务管理、资源调度、监控告警都依赖专有云平台自己的控制台和 OpenAPI。V3.16.0 这个版本号意味着你面对的是一个特定发行版接口形态、参数格式、支持的同步链路都可能和公共云文档有出入——这恰恰是开发指南存在的意义。适合谁读这篇一类是要在专有云上通过 API 批量创建 DTS 迁移或同步任务的人另一类是接了专有云项目、需要把 DTS 能力封装成内部平台或工具链的后端开发还有一类是运维同学——你需要知道任务出问题的时候从哪个接口查状态、看错误码、做重置。下面我按自己落地这类项目的经验把这份开发指南拆成能直接照着用的路径。2. 开发前必须摸清的三个底版本、Endpoint 和认证方式2.1 V3.16.0 到底对应哪套 API 语义专有云 Enterprise版的版本号跟公共云 OpenAPI 的版本号不是一回事。V3.16.0 是 Apsara Stack 平台的发行版本它决定了 DTS 服务暴露出来的 API 版本、参数校验规则和底层调度能力。我在项目中见过最典型的翻车现场拿公共云文档里的参数名直接往专有云接口上套结果创建任务时接口报 invalid parameter或者任务能创建但根本不调度。落地时你第一件事是确认专有云环境里 DTS 的 API 版本。常见做法是登录专有云控制台找到 DTS 服务详情页看服务版本号和 API 版本号。这个信息通常会出现在服务参数或关于页面里。V3.16.0 对应的 DTS OpenAPI 一般以版本号加日期区分比如 2019-01-01 这类。这个日期版本号才是你调用时 URL 里要带的别跟平台版本号搞混。提示如果你拿到的开发指南 PDF 或在线文档里写了 API 版本优先以它为准。版本号对不上时后面所有签名计算和参数格式都可能出错。确认版本还有个实际好处V3.16.0 里 DTS 支持的结构化迁移、全量迁移、增量同步、数据订阅这几个大能力API 路径是稳定的但参数细节在不同小版本里会有微调。比如迁移任务的源端实例类型在专有云里用的是SourceEndpoint.InstanceType取值包括RDS、ECS、LocalInstance等但专有云里还会多出ApsaraDB这种内部类型。这些细节只有对着 V3.16.0 的指南才能确认。2.2 Endpoint 与 AK/SK 的获取路径专有云的 DTS OpenAPI endpoint 不是公共云的dts.aliyuncs.com而是内网地址通常长这样dts-vpc.region.apsara-stack.domain或者dts.region.local。拿到准确 endpoint 的方法是查专有云平台的全局配置一般在控制台的「平台信息」或「服务地址」里能看到。我在做项目时习惯先把它配到环境变量里避免代码里到处硬编码。认证方式上专有云普遍兼容阿里云公共云的 AK/SK 签名机制——就是用 AccessKey ID 和 AccessKey Secret 对请求做 HMAC-SHA1 签名但签名的写法有几个专有云特有的坑。首先专有云的 endpoint 可能带端口号签名时 Host 头必须带上端口否则签名校验过不去。其次专有云里有些环境对 GMT 时间有严格要求服务器时钟偏差超过 15 分钟请求直接拒绝返回InvalidTimeStamp.Expired这类错误。import datetime import hashlib import hmac import base64 import urllib.parse # 专有云 DTS OpenAPI 签名示例V3.16.0 环境 access_key_id 你的AK access_key_secret 你的SK endpoint dts-vpc.cn-shanghai.apsara-stack.com action DescribeMigrationJobs params { Action: action, AccessKeyId: access_key_id, SignatureMethod: HMAC-SHA1, SignatureVersion: 1.0, SignatureNonce: 随机字符串, Timestamp: datetime.datetime.utcnow().strftime(%Y-%m-%dT%H:%M:%SZ), Format: JSON } # 按字典序排序并编码 sorted_params sorted(params.items()) query_string .join( f{urllib.parse.quote(k, safe)}{urllib.parse.quote(v, safe)} for k, v in sorted_params ) string_to_sign fGET%2F{urllib.parse.quote(query_string, safe)} signature base64.b64encode( hmac.new( access_key_secret.encode() b, string_to_sign.encode(), hashlib.sha1 ).digest() ).decode() url fhttp://{endpoint}/?{query_string}Signature{urllib.parse.quote(signature, safe)}这段代码里有个容易忽略的点SignatureMethod和SignatureVersion必须参与签名而且排序时按参数名的 ASCII 码排。专有云里常见的报错是SignatureDoesNotMatch八成是参数编码时把空格编成了而不是%20。另外注意AccessKeySecret后面要拼一个——这是阿里云签名算法的固定规则代表「仅 AK 密钥、无 Token」的场景。如果专有云环境开启了 STS 临时凭证那SecurityToken也要参与签名而且位置必须在排序后的参数字典里。2.3 专有云控制台与 API 的权限边界专有云的权限模型和公共云 RAM 基本一致但有个差异点专有云的子账号通常由平台管理员在本地做授权不一定能直接开 RAM 策略。我在某个集成项目里遇到的情况是运维给了我一对 AK/SK但调用CreateMigrationJob时返回NoPermission后来查出来是子账号只挂了 DTS 的只读权限没有写权限。开发前建议先确认三件事账号有没有 DTS 的读写权限DTS 服务的资源组Resource Group是否匹配目标实例比如 RDS的访问权限是否已经打通。专有云里 DTS 任务要读写源和目标实例靠的是 DTS 服务本身的账号体系不是你的 AK/SK 直连数据库。所以创建任务时你需要提供源库和目标库的连接信息这部分权限由 DTS 内部执行引擎处理跟你的 OpenAPI 权限是两码事。3. 用 API 创建并管理一个数据迁移任务核心流程与参数调优3.1 创建迁移任务从 CreateMigrationJob 到 ConfigureMigrationJob专有云 DTS OpenAPI 创建任务不是一步到位的。它分成两个阶段先用CreateMigrationJob拿到任务 ID再用ConfigureMigrationJob配置源库、目标库和迁移对象。我第一次做的时候以为一个接口全搞定结果任务创建完一直停在「未配置」状态查了指南才发现漏了第二步。这个设计有它的道理专有云平台需要先为任务分配资源、生成任务 ID然后才能开始配置链路。import requests base_url http://dts-vpc.cn-shanghai.apsara-stack.com headers {Content-Type: application/json} # 第一步创建迁移任务返回 MigrationJobId create_payload { Action: CreateMigrationJob, RegionId: cn-shanghai-apsara, MigrationJobClass: large, # 规格small/medium/large/xlarge SourceEndpoint.InstanceType: RDS, DestinationEndpoint.InstanceType: RDS, AccessKeyId: 你的AK, Signature: 上一步算出的签名 } resp requests.post(base_url, jsoncreate_payload, headersheaders) migration_job_id resp.json().get(MigrationJobId)拿到MigrationJobId后第二步配置任务。配置时需要注意MigrationJobClass决定迁移任务的并发度和性能上限专有云里常见取值是small、medium、large、xlarge。如果是几千万行的表全量迁移建议直接选large或xlarge否则全量阶段可能要跑十几小时。但你也要知道规格越大占用的调度资源越多专有云资源池有限同一个 region 下可能并发任务多了会排队。3.2 配置源库、目标库与迁移对象参数逐项说明配置任务这一步参数多且杂我把最关键的几组列出来参数组核心参数取值说明坑点源库SourceEndpoint.InstanceTypeRDS / ECS / LocalInstance专有云里 IDC 自建库通常用 LocalInstance源库SourceEndpoint.InstanceID实例ID或连接地址自建库填 IP:端口目标库DestinationEndpoint.InstanceTypeRDS / DRDS / MaxCompute确认目标实例类型和专有云版本匹配迁移对象MigrationObjectJSON 数组按库.表结构组织表名大小写敏感漏配会跳过同步初始化MigrationModeSTRUCT_INIT / FULL_INIT / INCREMENT三个值可组合用逗号分隔网络SourceEndpoint.IP自建库 IP专有云 DTS 需要能路由到该 IP# 第二步配置迁移任务节选关键参数 configure_payload { Action: ConfigureMigrationJob, MigrationJobId: migration_job_id, MigrationJobName: order_db_to_ods, MigrationMode: STRUCT_INIT,FULL_INIT,INCREMENT, SourceEndpoint.InstanceType: LocalInstance, SourceEndpoint.IP: 10.20.30.40, SourceEndpoint.Port: 3306, SourceEndpoint.UserName: dts_sync_user, SourceEndpoint.Password: 加密后的密码, DestinationEndpoint.InstanceType: RDS, DestinationEndpoint.InstanceID: rm-xxxxx, DestinationEndpoint.IP: 10.30.40.50, MigrationObject: [{\DBName\:\shop\,\TableIncludes\:[\orders\,\order_items\]}], Checkpoint: 0 # 增量同步的启动位点0 表示从当前时间启动 } resp requests.post(base_url, jsonconfigure_payload, headersheaders)注意MigrationObject是 JSON 字符串不是 JSON 对象。专有云 OpenAPI 对嵌套参数的处理跟公共云一致——所有参数都在顶层平铺数组类型用字符串表达。SourceEndpoint.Password不需要你做额外加密直接传明文DTS 内部会处理安全传输但示例里我习惯先做一层 Base64 避免日志把密码打出来——这纯属工程习惯不是协议要求。3.3 启动、暂停、停止与删除生命周期管理的正确姿势任务配置完成并不代表开始跑。你需要显式调用StartMigrationJob才会触发调度。这一步特别容易被漏掉因为有些公共云工具会在配置完自动启动但专有云 OpenAPI 不会。启动之后你就可以轮询DescribeMigrationJobs看状态了。# 启动任务 start_payload { Action: StartMigrationJob, MigrationJobId: migration_job_id, AccessKeyId: 你的AK, Signature: 上一步算出的签名 } requests.post(base_url, jsonstart_payload, headersheaders) # 轮询任务状态 describe_payload { Action: DescribeMigrationJobs, MigrationJobId: migration_job_id, PageSize: 10, PageNum: 1 } resp requests.post(base_url, jsondescribe_payload, headersheaders) status resp.json()[MigrationJobs][MigrationJob][0][MigrationStatus]专有云 DTS 任务状态机里有几个关键状态NotStarted已创建未配置、Prechecking前置检查中、Migrating迁移中、Suspending已暂停、Finished已完成、Failed失败。我一般用 10 秒间隔轮询连续查 5 次状态没变化就拉一次DescribeMigrationJobStatus看详情。注意Failed状态不代表任务终结比如网络抖动导致的连接失败你调整好网络后可以直接调StartMigrationJob再拉起来不用重建任务——这是 DTS 相对 DataX 这类离线工具的一个优势。3.4 重置与跳过增量同步卡住时的两个后悔药增量同步中经常遇到的问题是源库产生了一条大事务DTS 拉取超时报错任务状态变成Failed。这时候你有两个选择。如果只是单条数据有问题可以调SkipPreCheck或者直接在控制台跳过当前异常位点如果任务已经乱到没法继续就调ResetMigrationJob把任务重置到某个 checkpoint 位点重来。# 重置任务到指定时间点 reset_payload { Action: ResetMigrationJob, MigrationJobId: migration_job_id, Checkpoint: 2024-01-15T10:00:00Z } requests.post(base_url, jsonreset_payload, headersheaders)Checkpoint参数是时间字符串格式是 UTC 的 ISO8601。这里有个常见误解有人以为 checkpoint 可以填 binlog 文件名和位点但专有云 OpenAPI 里这个参数就是时间。如果你需要精确到 binlog 位点只能通过控制台的「重新同步」功能OpenAPI 不支持。做数据订正时我会先停止业务写然后重置到业务停止前一分钟这样增量追平后两边数据一致性最好。4. 结构迁移与实时同步的差异化配置从 MigrationMode 到同步延迟调优4.1 结构迁移只搬表结构不搬数据MigrationMode里的STRUCT_INIT表示只做结构迁移也就是把源库的表结构、索引、约束搬到目标库。这个能力在项目初期建影子库、做兼容性验证时特别有用。结构迁移的 API 参数跟全量迁移几乎一样区别只在MigrationMode的取值。但要注意结构迁移并不会迁移视图、存储过程、触发器等对象——至少 V3.16.0 的默认行为是这样。我在一个从 Oracle 迁移到 MySQL 的项目里用结构迁移先拉了 200 张表的 DDL 过去然后拿目标库的表结构和源库逐个对比。发现的问题包括Oracle 的NUMBER类型在 MySQL 里被映射成了DECIMAL(38, 0)精度丢失CLOB被映射成了LONGTEXT但没有处理字符集差异。所以结构迁移跑完别急着开始数据迁移先做一轮 DDL 差异比对是必须的。结构迁移的 API 调用不需要指定MigrationObject里的表列表那么细也可以按整个库迁移。但如果源库是 Oracle每个表的字段类型映射规则你最好先在测试环境验证一遍。专有云 DTS 的映射规则在开发指南的附录里有表我建议把那个表截下来贴在项目文档里团队做结构评审时很好用。4.2 全量增量组合如何设置 MigrationMode 达到最短停机时间生产环境做业务迁移时最怕的是停机窗口不够。DTS 的FULL_INIT,INCREMENT组合就是为这个场景设计的先全量搬完存量数据再通过增量同步追平迁移期间的新写入。这样你只需要在切流前停写几分钟让增量追平然后改连接串就行。# 全量增量组合的 MigrationMode 配置 MigrationMode: FULL_INIT,INCREMENT # 如果你只想全量一次不持续同步 MigrationMode: FULL_INIT但实际跑起来增量追平速度往往不如预期。常见原因是源库的写入量太大而 DTS 实例规格是small同步速率跟不上。这种情况我一般先看DescribeSynchronizationJobStatus里的Delay指标如果延迟持续增长说明增量消费能力不足。解决办法是把任务规格调大——但注意MigrationJobClass在创建任务时定死后期不能直接改。你要做的是新建一个xlarge规格的任务把 checkpoint 设成原任务当前位点然后切换过去。这个过程比较复杂我一般建议在创建任务前就按源库峰值写入量的 3 倍估算规格。4.3 同步延迟调优的 4 个参数除了规格之外有几个 OpenAPI 参数会影响同步延迟第一个是SourceEndpoint.DatabaseName如果源库是多个库尽量拆分任务避免一个大任务里串行拉取多个库。第二个是目标端的写入并发专有云 DTS 里可以通过DestinationEndpoint.MaxParallel调整目标库写入线程数但我遇到的场景里这个参数不是所有版本都开放。第三个是源端的拉取批次大小这个没有直接 OpenAPI 参数但可以通过设置Checkpoint的粒度间接控制。第四个是网络参数专有云 DTS 和源库之间如果有防火墙或安全组TCP 长连接数受限也会导致延迟上不去。注意调参之前先看延迟的瓶颈在源端还是目标端。如果目标库是 RDS观察它的 CPU 和连接数如果目标库是自建 MySQL看show processlist里来自 DTS 的会话是否长时间处于Waiting for table metadata lock。很多时候不是 DTS 慢是目标库写不进去。4.4 数据订阅模式从 OpenAPI 拉取增量变更除了迁移和同步V3.16.0 的 DTS 还支持数据订阅Data Subscription。这个能力和 Canal 类似但由 DTS 统一管理位点和消费组。OpenAPI 上需要先创建订阅任务然后创建消费组再用 SDK 连上去消费。# 创建订阅任务 subscribe_payload { Action: CreateSubscriptionJob, RegionId: cn-shanghai-apsara, SourceEndpoint.InstanceType: RDS, SourceEndpoint.InstanceID: rm-xxxxx, SubscriptionInstanceNetworkType: classic, SubscriptionObject: [{\DBName\:\shop\,\TableIncludes\:[\orders\]}] }订阅任务的消费端不走 HTTP OpenAPI而是用 DTS 提供的 Java SDK 建立长连接。专有云环境里这有一个特殊地方SDK 连接地址不是公网而是专有云的内部接入点一般形如dts-cn-shanghai.apsara-stack.com:8080。创建订阅任务后你需要从返回结果里拿SubscriptionInstanceId然后用 SDK 去订阅这个 ID 的消费组。这个流程和公共云文档差异较大我劝你一定以 V3.16.0 开发指南里的 SDK 版本为准不要从 Maven 仓库随手拉最新版——专有云 SDK 常常是老版本 fork 出来的新版的 Jackson 依赖可能冲突。5. 专有云 DTS 开发避坑5 条真实踩坑记录5.1 现象签名总是报SignatureDoesNotMatch原因是我在拼接待签名字符串时把 URL 里的参数值做了两次 percent-encode而 DTS 服务端只 decode 了一次。后来发现专有云的 API 网关对签名串的编码规则和公共云一致参数名和参数值都按 RFC 3986 编码空格编成%20而不是但很多人会在代码里顺手用 Python 的urlencode函数它默认把空格编成这就完了。解决方法是自己写一个编码函数用urllib.parse.quote并传入safe把所有保留字符都编码掉。签名计算用的是编码后的字符串但最终请求 URL 里也要用同样的编码串两边保持一致。如果你用 Java 或 Go 重写签名逻辑注意不要依赖 HTTP 客户端的自动编码因为会自动做一层 decode。5.2 现象任务创建成功但一直停在NotStarted后来查了开发指南才发现CreateMigrationJob这个接口在专有云里创建任务之后默认是「未配置」状态必须调ConfigureMigrationJob才能进入可启动状态。而且ConfigureMigrationJob里的参数如果缺了DestinationEndpoint或MigrationObject接口会直接报错任务状态更不会往前推。解决方法是把两步串成事务式调用——先配置成功再启动。我做了一个重试封装配置失败时调用DeleteMigrationJob清掉这个任务避免留下一堆坏任务占资源。专有云控制台上会看到这些未配置的任务数量多了运维会找你。5.3 现象全量迁移完成后增量同步延迟持续增长这个问题的根源通常不在 DTS 本身。我在一个从自建 MySQL 到 RDS 的同步任务里延迟一直涨到 120 秒。排查下来发现是源库开启了binlog_formatMIXEDDTS 为了兼容 MIXED 格式解析速度下降。另外源库有大量无主键表DTS 增量拉取时无法走索引定位只能全表扫描修改行效率极低。解决的思路一是给源库无主键表加上主键但生产环境加主键有风险要挑业务低峰期二是把binlog_format改成ROWDTS 对 ROW 格式解析效率最高。还有一招是调整 DTS 的MigrationJobClass到xlarge临时扩大增量消费能力等追平后再降下来。但降规格这个操作在 OpenAPI 里没有直接的修改接口仍然只能新建任务——所以一开始别贪小规格。5.4 现象目标库死锁导致任务失败大量并发写入目标库时如果目标库没有合适的索引DTS 的写入线程会触发死锁。报错形如Deadlock found when trying to get lock。原因是 DTS 默认对迁移任务是多线程并发写入同一个表如果表上没有唯一索引多个线程同时插入相似数据时容易互相阻塞。解决方法是先做结构迁移时给目标表补上唯一索引如果业务允许在迁移期间先去掉目标库的外键约束迁移完再恢复。另外可以把目标端的写入并发调低——DestinationEndpoint.MaxParallel设为4或8虽然慢一点但稳定。我在生产任务里一般先设8观察目标库锁等待情况再逐步调大最大到16。5.5 现象任务失败后重启报InvalidCheckpoint这通常是因为源库的 binlog 在任务暂停期间被清理了。专有云 RDS 的 binlog 保留时间默认可能只有 7 天如果你的任务停了超过这个时间重启时 DTS 找不到 checkpoint 对应的 binlog 文件。解决方法是确认 binlog 保留时长。对于 RDS 实例你可以在 RDS 控制台调整 binlog 保留周期对于自建库注意磁盘空间允许的话把expire_logs_days调大。但如果 binlog 已经被清了就没有后悔药只能重置任务重新做全量增量。这个坑我踩过之后现在做迁移项目都会先确认源库 binlog 保留策略再把 DTS 任务的延迟告警阈值设成 5 分钟超过就告警绝不拖到 binlog 被清。6. 进阶把 DTS OpenAPI 封装成内部迁移平台的关键技巧做到这一步你已经能通过 OpenAPI 管理单个迁移任务。但真实企业里DTS 往往要服务于几十甚至上百条数据链路手工调用 API 根本管不过来。我的做法是写一个轻量封装层把常用的「创建 配置 启动」串成一条命令同时加上状态机管理和告警通知。# 封装创建迁移任务的完整流程 class DtsClient: def __init__(self, ak, sk, endpoint): self.ak ak self.sk sk self.endpoint endpoint def create_and_start_migration(self, config): # 1. 创建任务 job_id self._call(CreateMigrationJob, { MigrationJobClass: config[job_class], SourceEndpoint.InstanceType: config[source_type], DestinationEndpoint.InstanceType: config[dest_type] }).get(MigrationJobId) # 2. 配置任务 self._call(ConfigureMigrationJob, { MigrationJobId: job_id, MigrationMode: config.get(mode, FULL_INIT,INCREMENT), SourceEndpoint.IP: config[source_ip], SourceEndpoint.Port: config[source_port], SourceEndpoint.UserName: config[source_user], SourceEndpoint.Password: config[source_password], DestinationEndpoint.InstanceID: config[dest_id], MigrationObject: config[migration_object] }) # 3. 启动任务 self._call(StartMigrationJob, {MigrationJobId: job_id}) return job_id封装层里最值得做的是「配置回滚」。如果ConfigureMigrationJob失败自动把已创建的任务DeleteMigrationJob清理掉不留垃圾任务。第二个值得做的是状态缓存——启动时先拉一遍现有的迁移任务列表缓存 job_id、状态、目标库这样你后续做批量巡检时不用每次都全量查接口。关于验证方法我常用「双跑比对」让 DTS 和 DataX 同时迁移同一张表到两个不同的目标库跑完之后用 SQL 做行数对比和校验和对比。DTS 的增量同步没有内置数据校验接口至少 V3.16.0 的 OpenAPI 里我没找到所以要在业务层写一个比对任务定时抽样对比源库和目标库的数据。把比对任务挂在运维监控里比 DTS 自己的状态更靠谱。最后一个技巧是给 DTS 封装层加上「幂等」。专有云 API 在网络超时后重试时同一个请求可能被执行两次导致创建出重复任务。我的做法是每次请求带一个幂等令牌放在SignatureNonce参数里服务端如果用同一个 nonce 会直接返回已处理过的结果。这样即使 HTTP 超时重试也不会产生重复任务。做专有云 DTS 开发核心思路就是把「对着文档查接口」变成「用封装层管任务」把「人工盯监控」变成「脚本轮询加告警」。希望这些经验帮你在 V3.16.0 环境里少踩几个坑。本文还有配套的精品资源点击获取