简介《阿里云专有云Enterprise版V3.16.0数据传输服务DTS开发指南》面向需要在专有云环境中集成DTS能力的企业开发者与架构人员重点解决数据迁移、同步及订阅任务的API接入与SDK调用问题。文档从快速入门和准备工作讲起说明AccessKey、STS凭证、公共Header参数及Endpoint获取方式并给出Java、Python、Go三类SDK的安装、身份验证、发起调用与错误处理示例API参考部分覆盖创建实例、配置迁移/同步/订阅任务、启动与批量启动、查询任务详情和日志、消费组管理、修改同步对象及账号密码等接口。资源包为1个PDF文件大小3.57MB目录按入门、准备、SDK示例和API参考组织便于按模块检索。已有75人学习适合具备一定接口开发基础、需要查阅参数说明和调用流程的读者作为案头参考。1. 从一次跨机房数据割接说起这份 DTS 开发指南到底解决什么问题上个月帮一个团队做专有云内的数据割接库表清单有七十多张业务方只给了一个两小时的窗口。手工在控制台上点“新建迁移任务”点完第三张表我就知道这条路走不通——不是点不完是点完之后没法审计、没法回滚、也没法复跑。真正能扛住这种场景的是把数据传输服务 DTS 的 API 接口开发接进自己的调度系统让建实例、配对象、启动、轮询状态全部变成代码。这份 V3.16.0 的开发指南讲的就是这条链路Java、Python、Go 三套 SDK 的调用姿势加上迁移、同步、订阅三类任务的完整接口清单。适合已经在专有云里跑业务、需要把数据迁移做成可重复流程的后端和运维开发如果你只是想迁一张表控制台比读文档快。2. 打通调用链POP 网关下的鉴权与四个必带 Header专有云企业版从 V3.16.0 起API 默认走 POP 网关调用体验和公有云对齐。早期版本走的是另一套内部网关文档里也留了说明老环境还能用但新写的代码没必要再迁就。整条链路是这样的你的程序拿 Endpoint 拼请求请求先到网关做签名校验和路由再落到 DTS 后端。中间任何一环参数缺失返回的错误都不一定指向真正的问题——这点后面踩坑章节会细说。先把一次成功调用需要凑齐的东西列清楚缺一样都跑不通要素从哪来缺了会怎样Endpoint控制台的 API 与工具页面按地域查域名解析失败或 404产品名 API 版本API 目录里搜接口名页面上会带出来网关找不到 ActionAccessKey 或 STS 三元组个人中心的密钥页 / 角色扮演接口签名校验失败RegionId公共 Header 参数页路由到错误地域组织 ID组织列表接口查询结果为空但不报错资源集 ID资源集列表接口默认为空资源隔离失效2.1 长期密钥与临时凭证别混着用AccessKey 是一对 ID 和 Secret用对称加密给请求做签名证明“这个请求是我发的”。它分两种来源个人账号的 AccessKey 和组织级的 AccessKey。组织级权限大得多只有运营管理员和一级组织管理员能取一般不建议塞进自动化脚本里——一旦泄漏影响面是整个组织。线上调度我一般强制走 STS也就是临时安全令牌。它由 AccessKey ID、AccessKey Secret 和 SecurityToken 三元组构成三个值缺一不可尤其是 SecurityToken很多人只传了前两个签名能算出来但网关直接拒绝。STS 的好处是有效期短、权限可以按角色收窄缺点是过期就得重新 AssumeRole调度系统里必须有一层凭证刷新逻辑否则跑到一半所有请求集体 403。提示个人账号 AccessKey 是受权限体系管控的受限凭证用它调接口时 Header 里必须补全地域、组织、资源集这几个限制参数否则很容易收到“权限不足”而实际原因只是 Header 没带全。2.2 三个必带 Header 和一个容易忘的实例 ID组织 IDx-acs-organizationid和资源集 IDx-acs-resourcegroupid这两个 Header是专有云和公有云最大的差别。不指定组织 ID接口默认取当前用户所属组织不指定资源集 ID默认为空。问题在于——这两个参数缺失时不会报错你只会看到一个空列表或者一个不属于预期资源集的任务。这种“静默失败”比报错恶心十倍我第一次遇到时对着日志查了两个小时。x-acs-regionid是地域x-acs-instanceid是具体实例 ID后者在部分接口里用于二次校验。四个参数里前三个建议做成 SDK 的全局配置一次配好全链路生效别在每个请求里手写。2.3 先写一个连通性自检再写业务代码血泪经验不要一上来就写创建实例的代码。先花二十分钟写个自检脚本把凭证、Endpoint、Header 三件事验证掉后面所有问题都会好排查得多。# probe_dts.py —— 只读探针验证鉴权链路是否打通 import os from dts_sdk import Client, Config # 包名按你环境里实际的 SDK 引入路径替换 from dts_sdk.models import DescribeDtsJobsRequest # 全部从环境变量取避免密钥进代码库 AK_ID os.environ[DTS_AK_ID] AK_SEC os.environ[DTS_AK_SECRET] TOKEN os.environ.get(DTS_STS_TOKEN) # 用长期 AK 时留空即可 ENDPOINT os.environ[DTS_ENDPOINT] # 形如 dts.cn-region-*.内部域名 cfg Config( access_key_idAK_ID, access_key_secretAK_SEC, security_tokenTOKEN, # 三元组的第三个值别漏 endpointENDPOINT, ) cfg.region_id os.environ[REGION_ID] # 组织 / 资源集通过额外 Header 透传这是专有云的关键差异点 cfg.extra_headers { x-acs-organizationid: os.environ[ORG_ID], x-acs-resourcegroupid: os.environ[RG_ID], } client Client(cfg) req DescribeDtsJobsRequest(page_number1, page_size1) # 只取一条成本最低 resp client.describe_dts_jobs_with_options(req, {}, {}) print(RequestId:, resp.body.request_id) # 出问题拿这个找运维 print(Total:, resp.body.total_record_count)这段逻辑的重点不在“查到了什么”而在“能不能查”。request_id是排查时的唯一凭据网关侧和后端侧的日志都靠它对上号所以每次调用都应该落到结构化日志里。参数上page_size设为 1 是为了压到最小开销security_token传空字符串和传None在部分 SDK 版本里行为不一致用长期 AK 时建议显式传None。如果这一步就失败按这个顺序看域名能不能解析 → 报错里有没有SignatureDoesNotMatch密钥错→ 有没有InvalidAccessKeyId.NotFoundAK 不存在或已禁用→ 有没有 403 但错误码很含糊多半是组织 ID 或资源集 ID 不对。这四步走完九成环境问题都能定位。3. Python SDK 跑通迁移任务从建实例到启动的四段调用文档里 Java、Python、Go 三套示例的结构是一致的安装、设凭证、配连接、发起调用、错误处理最后还有一个 CommonRequest 兜底方案。选择哪套语言不重要重要的是理解调用顺序——DTS 的任务生命周期是严格分段的跳步一定失败。3.1 装包与凭证注入的三种姿势安装本身没什么可说的专有云环境通常没有外网走内部源# 从内部镜像源装版本号按你环境的制品库为准 pip install dts-sdk-package -i https://内部源/simple --trusted-host 内部源 # 凭证走环境变量别写在代码或配置文件里 export DTS_AK_IDak-id export DTS_AK_SECRETak-secret export DTS_STS_TOKENsecurity-token # 仅 STS 模式需要 export DTS_ENDPOINTdts.cn-region-*.内部域名 export REGION_IDcn-region-* export ORG_ID组织ID export RG_ID资源集ID凭证注入有三种常见做法硬编码绝对不要、配置文件要控制文件权限、环境变量或密钥管理服务推荐。容器化部署时第三种最省事配合编排平台注入 Secret 就行。要注意的是环境变量对子进程可见如果你的调度系统会在同一台机器上跑别人的任务这一点得评估。3.2 三段式调用建实例、配对象、启动DTS 的任务不是一个接口建完的。先创建实例再给实例配置迁移或同步对象最后启动。中间还有一步预检查可以显式调用也可以隐式触发。下面是骨架代码字段语义我保留完整具体字段名以你手上 API 目录里的定义为准——不同小版本之间偶有差异照抄容易翻车。# create_and_start.py import os, json, time from dts_sdk import Client, Config from dts_sdk.models import ( CreateDtsInstanceRequest, # 创建实例 ConfigureMigrationJobRequest, # 配置迁移对象 StartDtsJobRequest, # 启动任务 DescribeDtsJobDetailRequest, # 查详情 ) cfg Config( access_key_idos.environ[DTS_AK_ID], access_key_secretos.environ[DTS_AK_SECRET], security_tokenos.environ.get(DTS_STS_TOKEN), endpointos.environ[DTS_ENDPOINT], ) cfg.region_id os.environ[REGION_ID] cfg.extra_headers { x-acs-organizationid: os.environ[ORG_ID], x-acs-resourcegroupid: os.environ[RG_ID], } client Client(cfg) # ---- 第 1 段创建实例拿到 InstanceId ---- create_req CreateDtsInstanceRequest( regionos.environ[REGION_ID], typeMIGRATION, # MIGRATION / SYNC / SUBSCRIBE instance_classsmall, # 规格按数据量和并发选 pay_typePayAsYouGo, # 按量付费跑完释放最省 auto_payTrue, # 关键不置 True 会卡在待支付 quantity1, ) create_resp client.create_dts_instance(create_req) instance_id create_resp.body.instance_id print(instance:, instance_id) # ---- 第 2 段配置迁移对象结构与全量/增量开关都在这 ---- migration_object { DBList: { app_db: { # 源库名 TableList: {orders: {ColumnList: *}, users: {}}, NewSchemaName: app_db, # 目标库改名时用 } } } cfg_req ConfigureMigrationJobRequest( migration_job_idinstance_id, migration_mode{ StructureIntialization: True, # 结构初始化 DataIntialization: True, # 全量数据 DataSynchronization: True, # 增量同步 }, migration_objectjson.dumps(migration_object), source_endpoint{ # 源端连接信息 InstanceType: RDS, EngineName: MySQL, IP: 源实例地址, Port: 3306, UserName: 账号, Password: 密码, }, destination_endpoint{ # 目标端连接信息 InstanceType: RDS, EngineName: MySQL, IP: 目标实例地址, Port: 3306, UserName: 账号, Password: 密码, }, ) client.configure_migration_job(cfg_req) # ---- 第 3 段启动注意预检查告警的确认开关 ---- start_req StartDtsJobRequest( dts_job_idinstance_id, block_confirmationTrue, # 预检查有告警时仍强制启动慎用 ) client.start_dts_job(start_req)逻辑说明分三段看。第一段的auto_pay是最容易漏的字段按量付费实例不置True会停在未支付状态后面配置对象时直接报实例不可用。第二段的migration_mode三个布尔值决定这条任务是只做结构、只做全量、还是持续增量三个都开就是最常见的“结构全量增量”组合migration_object是 JSON 字符串注意ColumnList写*表示全列要裁剪字段就写成具体的列名列表。第三段的block_confirmation是把双刃剑预检查有告警时置True能强行启动代价是把风险从接口层挪到了你的业务层。参数上再补一句instance_class不是越大越好规格影响的是同步并发和内存缓存小规格跑大表会频繁触发内存告警反而拖慢速度。我的经验是按单表最大行数和日增量估算先小规格试跑一条增量看延迟曲线再决定要不要升配——接口里也提供了升降配能力不必重建任务。3.3 CommonRequestSDK 没覆盖到的接口怎么办专有云版本的 SDK 更新往往滞后于 API 目录经常出现“文档里有这个接口SDK 里没有对应方法”。这时候不要急着降级或者绕路用通用请求CommonRequest直接打 Action 就行# 用 CommonRequest 调用 SDK 未封装的接口 from dts_sdk.common import CommonRequest req CommonRequest() req.set_domain(os.environ[DTS_ENDPOINT]) # Endpoint req.set_version(2018-xx) # API 版本去 API 目录对照 req.set_action_name(ConfigureSubscriptionInstance) req.set_method(POST) req.add_query_param(SubscriptionInstanceId, instance_id) req.add_query_param(SubscriptionObject, json.dumps(sub_object)) resp client.do_action_with_exception(req)三个参数必须和 API 目录里完全一致set_version的版本串、set_action_name的动作名、以及每个 query 参数的键名。大小写错了不会给你“参数名写错”的提示只会返回InvalidParameter然后你得逐个对照。用 CommonRequest 调通的接口等 SDK 升级后再换回标准方法保留原来的请求构建逻辑可以当回归用例。3.4 返回值怎么读所有响应里第一眼要看的不是业务数据是RequestId。成功与否看 HTTP 状态码结合错误码字段业务侧还常见Success布尔值。异常处理不要只print(e)把错误码、错误消息、RequestId三个都提出来try: resp client.start_dts_job(start_req) except Exception as e: # 错误码分类处理别一律重试 code getattr(e, code, Unknown) if code in (Throttling, ServiceUnavailable): time.sleep(3) # 限流和短暂不可用可重试 elif code in (InvalidParameter, Forbidden): raise # 参数和权限问题重试无意义 else: raise错误码大致分四类参数类InvalidParameter、MissingParameter、权限类Forbidden、InvalidAccessKeyId、限流类Throttling、服务端类InternalError、ServiceUnavailable。前两类重试一万次也没用后两类才值得退避重试而且要有上限。这条规则写在调度框架里比写在每个业务函数里靠谱。4. 任务参数怎么配迁移、同步、订阅三类对象的字段对照三类任务的对象结构长得像但字段语义差别不小混用会得到一堆看不懂的报错。下面这张对照表是我从文档的“参数补充说明”章节和实际调用里整理出来的可以当速查用。维度迁移对象 MigrationObject同步对象 SynchronizationObject订阅对象 SubscriptionObject核心结构DBList → TableList → ColumnList同左但支持更细的 DDL/DML 过滤只到库表不含列裁剪是否支持改名支持目标库/表重命名支持不支持增量能力有靠 DataSynchronization 开关有且可配置冲突处理策略本质就是增量订阅典型用途上云迁移、跨库搬迁双写过渡、异地多活下游缓存刷新、数仓实时接入常见坑大小写敏感模式全镜像匹配开关消费位点与消费组4.1 三类对象的构造要点迁移对象的关键在DBList的嵌套层级库名 → 表名 → 列名逐层收敛。只迁部分列的时候ColumnList写成列名数组不写就是全列。如果源库和目标库同名表结构不一致NewSchemaName和表级的新表名映射就要用上否则预检查的结构一致性检查会直接报差异。同步对象和迁移对象最大的区别是多了“冲突处理”和“全镜像匹配”这类开关——同步是长期运行的源端一条 UPDATE 没带主键条件就可能把目标端整表覆盖。文档里专门提供了设置全镜像匹配开关和查询开关状态的接口这条链路的正确做法是先查当前开关状态确认符合预期再启动作业而不是默认它是关的。订阅对象最简单但也最容易出事。订阅链路不是“启动就完事”还需要管理消费组和消费位点。文档里给了四个消费组接口新增、查询详情、修改、删除加上一个修改消费位点的接口。典型的自动化流程是配置订阅通道 → 购买订阅实例 → 启动 → 建消费组 → 下游消费 → 需要回放时修改位点。少了消费组下游根本拉不到数据。4.2 预检查不是可选项预检查接口返回的是一组检查项每项带状态和说明。常见的告警包括源库 binlog 格式不对、目标库存在同名对象、账号权限不足、字符集不一致。很多人图快直接用block_confirmation强启结果全量跑完了增量追不上或者结构初始化报主键冲突。我一般的做法是把预检查结果做一次解析按严重程度分三档阻断项必须修、告警项记录并知会业务方、提示项忽略。代码里对阻断项直接中断流程不允许强启只有告警项才允许带block_confirmation通过并且要在任务记录里留痕。文档里还有一个“是否屏蔽预检查告警项目”的接口这个别乱用屏蔽之后等于永久关掉了这块的防线。4.3 无效配置的早期识别参数错了有时不会立刻报而是跑到一半才暴露全量阶段正常切到增量就停在某个位点不动。这类问题在配置阶段就能识别一半——检查源库的日志保留时长、目标库的写入权限、以及migration_mode三个开关是否和你的目标一致。只做一次性搬迁却把增量打开任务永远不会“完成”只会一直追日志白烧规格。5. 避坑排查我在专有云 DTS 上翻过的五次车5.1 查询任务列表永远返回空现象接口返回 200RequestId正常TotalRecordCount是 0但控制台上明明有一堆任务。原因请求 Header 里漏了组织 ID。专有云的权限体系是分组织的不指定组织时默认取当前用户所属组织如果你的 AK 属于组织 A而任务建在组织 B自然查不到。这个过程完全不报错是设计使然。解决把x-acs-organizationid做成 SDK 全局配置且从配置中心读取而不是硬编码。资源集 ID 同理指定资源集时必须同时指定组织。上线前用两个不同组织的账号交叉验证一次确认隔离生效。5.2 STS 模式跑批到一半集体 403现象调度系统连续跑了四十分钟正常突然所有请求返回签名错误或权限不足。原因STS 三元组有有效期过期后SecurityToken失效而 SDK 客户端是长生命周期的单例凭证没有跟着刷新。解决在客户端外面包一层凭证提供者每次调用前检查剩余有效期低于阈值就重新 AssumeRole 并重建客户端。不要图省事把有效期设成很长那等于把临时凭证变成长期凭证失去了用它的意义。5.3 任务启动成功但数据一直不落地现象StartDtsJob返回成功查详情状态也是运行中但目标表里一行数据都没有。原因八成是migration_mode的三个开关里全量那个没打开任务在等增量日志或者源 endpoint 的账号只有读权限拿不到 binlog。还有一种情况是对象配置里的库名大小写和实际不一致被全镜像匹配之类的开关吞掉了。解决先查任务详情的子任务状态看结构初始化、全量初始化、增量同步三个阶段的进度分别在哪一步停住。再看任务日志接口里面会有明确的位点信息和错误行。库表名一律用实际大小写拼写别凭记忆。5.4 重复触发导致多买了实例现象调度重试机制把创建实例的请求发了三次账单上多出三个按量实例。原因创建实例不是幂等操作网络超时重试时会重复执行。这是分布式系统的老问题接口层面没法替你解决。解决调用侧加幂等键用业务单号做去重表创建前先按名字或标签查一次是否已存在。同时给创建接口单独设重试策略——只重试明确的限流和服务端错误超时一律当作“结果未知”先查后建。释放同样要幂等批量释放接口能一次清多个但要确认任务已经结束运行中的任务释放不掉。5.5 任务结束但仍在计费现象迁移跑完了状态是已完成但资源还在。原因完成和释放是两个动作。任务完成后实例仍然占着规格按量付费的一直在算钱。解决调度流程的最后一步固定是“结束任务 → 释放实例”两步都要调且要校验返回。批量接口可以一次处理多条但要先过滤出已结束状态的任务否则整批失败。这条我写进了流程模板做完必须核对一次实例列表确认没有残留。6. 把调用包进调度系统状态轮询、幂等与验收位点单次调用跑通只是及格线真正决定这套东西能不能用的是把它塞进调度系统之后的表现。三个动作值得单独设计。第一是状态轮询的节奏。不要固定间隔死轮询全量阶段任务可能跑几十分钟增量阶段可能是几小时用指数退避加最大间隔封顶起始 5 秒每次乘 1.5封顶 60 秒。轮询接口本身也要看成本查列表比分页查详情便宜先查列表拿到状态只有状态变化时才查详情。第二是幂等。前面踩坑章节说过创建实例不幂等其实“配置对象”和“启动”同样不幂等——对已启动的任务再次调用启动部分版本会报错部分会静默重置。稳妥的做法是本地维护一张任务状态表每次调用前先对齐远端状态远端状态和本地预期不符就人工介入而不是盲目重试。第三是验收。任务显示“已完成”不代表数据对得上。我一般会拿三个指标做交叉验证源和目标的行数比对、几个关键表的校验和、以及增量位点是否追平。位点追平最容易被忽略——全量完成但增量延迟还在涨说明下游写入能力跟不上这时候任务状态是绿的数据是错的。最后一个我在流程模板里加的步骤是“跑完必须调一次任务日志接口把最后 100 行落盘归档”因为出问题时日志保留期往往已经过了这份归档就是唯一的后悔药。轮询阶段建议间隔观察指标结构初始化5 秒是否卡在 DDL 失败全量初始化15 到 30 秒已完成行数增速增量同步30 到 60 秒延迟位点是否收敛结束后一次实例是否已释放从那次七十多张表的割接以后我每次接数据迁移的活都强制先跑一遍自检脚本再走“预检查 → 分档处理 → 启动 → 位点验收”四步哪怕只迁一张表也不跳过。这套流程救过我至少两次——一次是组织 ID 配错一次是增量打开但 binlog 保留期只有一小时。希望帮到你。本文还有配套的精品资源点击获取