简介阿里云专有云Enterprise版V3.16.0数据传输服务DTS开发指南面向企业开发者、系统集成商及运维人员解决专有云环境通过API完成数据迁移、同步与订阅任务建设问题。压缩包内为单个PDF文件约3.57MB可离线查阅覆盖法律声明、通用约定、快速入门、准备工作、SDK示例与API参考等章节。准备工作含登录控制台、查看API信息、获取AK/SK与STS令牌、公共Header参数及DTS的EndpointJava、Python、Go的SDK示例覆盖安装、鉴权、请求配置、调用和错误处理。API参考覆盖创建DTS实例、配置迁移/同步/订阅任务、启动与批量启动、消费组管理、查询与修改任务等接口逐项说明参数、请求方式与响应信息是开发对接与排错依据。当前已有71人学习下载适合专有云DTS接口开发团队参考。1. 专有云 DTS 开发先对照 V3.16.0 这份指南把弯路走直很多从公有云切到专有云Apsara Stack的同事第一反应都是拿公有云的 DTS OpenAPI 文档直接怼结果在 Endpoint 和鉴权上卡了一整天。阿里云专有云 Enterprise版 V3.16.0 的数据传输服务 DTS 开发指南其实就是官方给的一张“地图”——它明确了在专有云环境下调用 DTS 接口的完整路径从 POP 网关的调用流程、AccessKey 和 STS 的获取到 Java、Python、Go 三种 SDK 的接入再到数据迁移、同步、订阅三大类 API 的参数定义全部收在了一本里。适合谁两类人一类是要把 DTS 的迁移/同步能力封装进自家运维平台的后端开发另一类是给客户做专有云交付、需要写自动化脚本的交付工程师。这篇笔记我会照着实际拆项目的方式把从准备到调通的完整链路和踩过的坑一次说清。2. 开工前的准备工作AccessKey、Endpoint 和公共 Header 的细节2.1 登录 API 与工具控制台获取 AccessKey 的两种路径专有云环境里没有公有云那种“AccessKey 管理”的独立入口而是统一收口在 Apsara Uni-manager 运营控制台里。首件事是先找部署人员要到运营控制台的访问域名然后用 Chrome 登录。这里有个细节首次登录会强制改密码密码要求 10~32 位且必须包含大写字母、小写字母、数字、特殊符号中的至少两类。如果账号开了 MFA登录流程会多一步绑定虚拟 MFA 设备6 位动态码输错三次会被锁这个在交付现场很常见。登录之后获取 AccessKey 分两条路个人账号 AccessKey右上角头像 - 个人信息 - 阿里云 AccessKey 区域直接查看。这种方式拿到的 AK 是受限的调用 API 时必须在 Header 里补x-acs-regionid、x-acs-organizationid、x-acs-resourcegroupid甚至x-acs-instanceid否则会报权限不足。组织 AccessKey只有运营管理员和一级组织管理员能操作路径是企业 - 资源管理 - 组织管理 - 目标一级组织 - 管理 Accesskey。组织级 AK 权限更大但操作风险也高一般不建议在业务代码里直接用。我一般建议业务代码里优先使用个人 AK 配合 STS 临时凭证权限可控泄露了也能快速失效。2.2 获取 STS AccessKey三元组缺一不可如果走 STS 方式流程是先拿到 RAM Role。在运营控制台右上角头像 - 个人信息 - 查看当前角色策略里可以看到当前的RAM Role。拿到角色后用 STS 的 AssumeRole 接口换取临时凭证得到三元组AccessKey ID、AccessKey Secret、SecurityToken。# 使用 alibabacloud_sts20150401 SDK 获取临时凭证 from alibabacloud_sts20150401.client import Client from alibabacloud_sts20150401 import models as sts_models from alibabacloud_tea_openapi.models import Config config Config( access_key_idLTAI5tXXXXXXXXXX, # 主账号 AK access_key_secretXXXXXXXXXXXXXXXXXXXX, # 主账号 SK endpointsts.aliyuncs.com # 专有云环境请用内部 STS Endpoint ) client Client(config) req sts_models.AssumeRoleRequest( role_arnacs:ram::1234567890:role/dts-role, # RAM Role role_session_namedts-dev-session # 自定义会话名 ) resp client.assume_role(req) # 三元组从这里取 temp_ak resp.body.credentials.access_key_id temp_sk resp.body.credentials.access_key_secret security_token resp.body.credentials.security_token这段代码的要点在role_arn和role_session_name。role_arn是准备阶段拿到的 RAM Role 完整 ARNrole_session_name是临时会话名最终会体现在操作审计日志里。换出来的SecurityToken在调用 DTS 接口时必须放在 Header 的x-acs-security-token里SDK 通常封装了这块但如果你直接用原生 HTTP 调用漏掉这个字段就会一直报InvalidAccessKeySecret。2.3 公共 Header 参数组织和资源集隔离是最大差异点专有云 V3.16.0 的 DTS 接口和公有云最大的差异就是多了组织和资源集的概念。调用 API 时需要在 Header 里带两个参数否则会碰到“查不到任务”或者“没有权限操作”的玄学问题参数名是否必填说明x-acs-regionid推荐地域 ID在专有云里通常形如cn-hangzhou-*具体值需要从运维平台确认x-acs-organizationid推荐组织 ID通过运营控制台开发指南里的GetOrganizationList接口获取x-acs-resourcegroupid可选资源集 ID通过ListResourceGroup获取。注意不指定则默认为空但如果指定了资源集就必须同时指定组织 IDx-acs-instanceid按需实例维度权限限制时使用这里有个容易翻车的地方如果你用的是个人账号 AK并且没传x-acs-organizationid接口可能不会直接报错而是返回一个空的任务列表。因为系统默认只查“当前用户所属组织”的数据而这个“所属组织”在未显式指定的情况下可能并不是你预期的那一个。排查这种问题最好的办法就是抓包看实际请求的 Header别上来就怀疑网络。2.4 获取 DTS 的 Endpoint历史版本和 V3.16.0 的差异文档里明确了一个关键变化自企业版 V3.16.0 开始专有云 API 默认使用 POP 网关方式调用而历史版本用的是 ASAPI 网关。这两种网关的 Endpoint 获取方式完全不同。POP 网关通过云产品开发指南里提供的 Endpoint 获取DTS 的 Endpoint 形如dts-pop.dts.aliyuncs.com专有云内部域名具体以实际环境为准。ASAPI 网关历史方式需要通过 ASAPI 的 Endpoint 再转发到具体云产品链路多一层性能和稳定性都不如 POP。拿到的 Endpoint 最后要配置到 SDK 的endpoint参数里Java 和 Python 的配置方式稍有不同下面会写。这里先记住一个原则如果是 V3.16.0 版本优先用 POP 网关如果是旧版本才考虑 ASAPI。在不确认版本的情况下先去运营控制台查版本号不要想当然。3. 用 Java 和 Python SDK 拉起第一个 DTS 任务3.1 Java SDK 的引入、凭证设置与连接配置Java 项目用 Maven 管理依赖的话在pom.xml里加入 DTS 的 SDK 依赖dependency groupIdcom.aliyun/groupId artifactIddts20200101/artifactId version1.0.0/version /dependency如果阿里云官方 Maven 仓库拉取慢可以在settings.xml里配置阿里云镜像。这个在专有云开发环境里尤其常见——很多内网机器访问外网 Maven 仓库受限配置私有 Nexus 或阿里云镜像几乎是必修课。然后初始化客户端import com.aliyun.dts20200101.Client; import com.aliyun.teaopenapi.models.Config; public class DtsClientFactory { public static Client createClient(String accessKeyId, String accessKeySecret, String endpoint) { Config config new Config() .setAccessKeyId(accessKeyId) .setAccessKeySecret(accessKeySecret) .setEndpoint(endpoint); // 例如dts-pop.dts.aliyuncs.com return new Client(config); } }这里有一个容易被忽略的点Config里还可以设置regionId但 DTS 的 Java SDK 对于地域的判断更多依赖 Header 里的x-acs-regionid所以光在 Config 里设置regionId是不够的需要在每个请求的 Header 里显式带上组织和资源集信息。后面发起调用的代码里我们会看到如何把公共参数塞进请求。3.2 Java 调用创建迁移任务的完整链路以“创建一个迁移任务”为例完整链路是购买实例 - 配置迁移任务 - 启动任务。这里贴出配置任务和启动的部分import com.aliyun.dts20200101.models.CreateMigrationJobRequest; import com.aliyun.dts20200101.models.CreateMigrationJobResponse; import com.aliyun.dts20200101.models.StartMigrationJobRequest; import com.aliyun.tea.TeaException; public class DtsMigrationDemo { public static void main(String[] args) { Client client DtsClientFactory.createClient( LTAI5tXXXXXXXXXX, XXXXXXXXXXXXXXXXXXXX, dts-pop.dts.aliyuncs.com); // 1. 配置迁移任务 CreateMigrationJobRequest createReq new CreateMigrationJobRequest(); createReq.setRegionId(cn-hangzhou-xxx); // 这里是关键专有云环境下源和目标实例信息都通过 JSON 字符串传递 createReq.setSourceEndpointInstanceType(RDS); createReq.setSourceEndpointInstanceId(rm-xxxxxxxx); createReq.setSourceEndpointEngineName(MySQL); createReq.setSourceEndpointRegion(cn-hangzhou-xxx); createReq.setDestinationEndpointInstanceType(RDS); createReq.setDestinationEndpointInstanceId(rm-yyyyyyyy); createReq.setDestinationEndpointEngineName(MySQL); // 迁移对象库名.表名 createReq.setMigrationObject([{\CanProduceSQL\:true,\DBName\:\test_db\,\TableIncludes\:[...]}]); createReq.setMigrationModeDataInt(true); // 全量迁移 createReq.setMigrationModeDataInt(false); // 增量迁移注意这里的命名 createReq.setMigrationReserved({\isSessionVerify\:true}); // 2. 发起请求捕获异常 try { CreateMigrationJobResponse createResp client.createMigrationJob(createReq); String jobId createResp.body.getJobId(); System.out.println(创建成功任务ID: jobId); // 3. 启动迁移任务 StartMigrationJobRequest startReq new StartMigrationJobRequest(); startReq.setJobId(jobId); client.startMigrationJob(startReq); System.out.println(迁移任务已启动); } catch (TeaException e) { // 打印错误码和具体信息专有云环境下 ErrorCode 往往是排查关键 System.err.println(错误码: e.getCode()); System.err.println(错误信息: e.getMessage()); } } }这段代码里有几个参数特别容易搞混MigrationModeDataInt和MigrationModeDataInt看起来相似但表达的意思完全不同——前者代表是否做全量数据迁移后者代表是否做增量数据同步。在配置迁移任务时很多人只开全量不开增量结果基础数据过去了线上业务数据一直在源库累积两边永远追不平。另外MigrationObject的格式是 JSON 数组字符串必须以[{}]开头里面DBName是库名TableIncludes是表白名单TableExcludes是表黑名单。3.3 Python SDK同样的逻辑更简洁的写法Python 的接入方式在思路上和 Java 完全一致区别在于 SDK 的安装和参数传递方式。安装依赖pip install alibabacloud_dts20200101然后看一个配置同步任务并启动的实例from alibabacloud_dts20200101.client import Client from alibabacloud_dts20200101 import models as dts_models from alibabacloud_tea_openapi.models import Config config Config( access_key_idLTAI5tXXXXXXXXXX, access_key_secretXXXXXXXXXXXXXXXXXXXX, endpointdts-pop.dts.aliyuncs.com ) client Client(config) # 配置同步任务这里的实例类型支持 RDS、POLARDB、DRDS 等 req dts_models.CreateSynchronizationJobRequest( region_idcn-hangzhou-xxx, source_endpoint_instance_typeRDS, source_endpoint_instance_idrm-xxxxxxxx, source_endpoint_engine_nameMySQL, destination_endpoint_instance_typeRDS, destination_endpoint_instance_idrm-yyyyyyyy, destination_endpoint_engine_nameMySQL, synchronization_job_namesync-job-001, structure_initializationTrue, # 结构初始化 data_initializationTrue, # 全量初始化 synchronization_directionForward # 同步方向Forward 为正同步 ) try: resp client.create_synchronization_job(req) job_id resp.body.parameters.get(SynchronizationJobId) print(f同步任务ID: {job_id}) # 启动同步 start_req dts_models.StartSynchronizationJobRequest( synchronization_job_idjob_id, synchronization_directionForward ) client.start_synchronization_job(start_req) print(同步任务已启动) except Exception as e: # Python SDK 的错误信息里一般包含 ErrorCode优先看这个 print(f调用失败: {e})这段代码里的structure_initialization和data_initialization对应控制台里的“结构迁移”和“全量迁移”。如果你只是做增量同步这两个参数都设False同时要在后台任务配置里指定同步起始位点。还有synchronization_direction这个参数双向同步时使用Inverse方向创建第二个任务单向同步只传Forward传反了任务必挂。3.4 当 SDK 缺少某个接口用 CommonRequest 兜底DTS 的 SDK 更新速度不一定跟得上 API 版本。偶尔会遇到文档里有某个接口但 Java/Python SDK 里没有对应方法的情况。这时候不用干瞪眼直接用各语言 SDK 里的 CommonRequest 发起原生请求import com.aliyun.teaopenapi.models.CommonRequest; import com.aliyun.teaopenapi.models.Config; import com.aliyun.dts20200101.Client; public class CommonRequestDemo { public static void main(String[] args) throws Exception { Client client DtsClientFactory.createClient(AK, SK, dts-pop.dts.aliyuncs.com); CommonRequest request new CommonRequest(); request.setVersion(2020-01-01); // API 版本号 request.setAction(DescribeDtsJobDetail); // 接口名 request.setProtocol(HTTPS); request.setMethod(POST); // 通过 query 传递接口参数 request.query.put(DtsJobId, xxxxxx); request.query.put(RegionId, cn-hangzhou-xxx); // 发起请求SDK 会自动处理签名 CommonResponse response client.common(request); String result response.getBody(); System.out.println(result); } }CommonRequest的用法核心在于setAction和query参数必须跟文档里的参数名完全一致大小写敏感。请求体返回的是 JSON 字符串建议直接用Map解析。这个方法在网络调试阶段也很有用——当怀疑 SDK 封装有问题时用 CommonRequest 裸调接口可以快速定位是我方参数问题还是服务端异常。4. DTS 开发避坑指南预检查、任务状态和报错排查4.1 预检查失败源库/目标库连接不通现象创建迁移或同步任务后状态一直停在“未启动”点击启动后立刻失败预检查详情里报“连接数据库失败”或“源库实例 ID 不存在”。原因专有云环境的网络隔离比公有云严格得多。DTS 服务访问源库和目标库依赖的是专有云内部的网络白名单和 VPC 路由。最常见的原因是白名单里没有放行 DTS 的 IP 段或者源库是自建库DTS 所在网段和源库网段路由不通。解决文档 6.8.4 专门提供了“查询 DTS 服务的 IP 地址”的接口先调用这个接口拿到当前环境 DTS 服务的完整 IP 列表然后把这些 IP 全部加到源库和目标库的白名单里。如果是自建库还要确认 DTS 所在交换机和数据库所在交换机之间有路由策略。这里不要凭经验猜专有云环境不同企业版的 IP 段差异很大查出来再放行最稳妥。4.2 任务状态长时间卡在“预检查”或“初始化”现象任务启动后状态一直显示“预检查中”超过 10 分钟或者“结构初始化”报告成功但“全量初始化”一直 0%。原因这两个状态卡住的原因通常不同。预检查卡住大概率是目标库有 DDL 锁比如有人在目标库执行了大事务初始化卡住如果结构初始化成功说明 DTS 到两端数据库的网络是通的大概率是源库存在无主键表DTS 做全量迁移时要走特殊机制速度会骤降。解决先到目标库执行SHOW PROCESSLIST杀掉长事务再重启任务。如果初始化一直 0%用文档里的“查询子任务执行状态”接口逐一核对定位到具体是哪张表卡住。对无主键表在迁移对象里排除掉或者提前给源表加上主键再迁移。4.3 使用个人 AccessKey 报“权限不足”现象AK 明明有权限调用接口返回Forbidden或NoPermission。原因个人 Account AccessKey 是受限的必须携带完整的 Header 参数。很多人只传了x-acs-regionid漏了x-acs-organizationid和x-acs-resourcegroupid。解决在调用任何 DTS 接口前先确认当前用户所属的组织 ID 和资源集 ID。这两个 ID 需要通过运营控制台开发指南里的GetOrganizationList和ListResourceGroup接口获取。拿到后在客户端全局设置统一注入 Header避免每个请求单独传。我在项目里习惯封装一个DtsApiClient基类在before_action里统一处理这些 Header后续每个业务方法都不需要再关心。4.4 查询任务列表返回为空现象用 API 查询 DTS 任务列表接口返回 200 但Items数组为空用同一个 AK 登录控制台却能看到任务。原因控制台界面默认查的是所有组织和资源集的任务但 API 如果不显式传x-acs-organizationid默认只取当前用户所属组织而且资源集 ID 为空。两边查询范围对不上自然就出现“控制台能看见、API 查不到”的诡异情况。解决查询任务列表时显式带上x-acs-organizationid和x-acs-resourcegroupid。如果任务是在默认资源集下创建的x-acs-resourcegroupid传空字符串即可如果创建时指定了资源集查询时也必须指定同一个资源集 ID否则永远查不到。4.5 同步延迟持续增大不追平现象数据同步任务状态正常但同步延迟从一开始的几秒涨到几分钟甚至几小时且一直没有下降趋势。原因专有云 DTS 的同步延迟和源库写入量、目标库规格、网络带宽都有关系但最常见的还是目标库出现了大批量慢查询或者同步对象里有大表没有做分片。解决查看目标库的慢查询日志确认是不是同步账号写入被阻塞。如果同步的库表特别多可以用文档里的“批量启动任务”接口把这些任务拆到多个 DTS 实例上分摊压力。另外同步任务支持升级规格文档 6.9.6 有专门的升级实例规格接口紧急情况下可以先升两级追平后再降回来。5. 从 API 到落地订阅任务、消费组与自动化运维5.1 查询 DTS 服务 IP 并配置安全组如果源库和目标库都在专有云 VPC 内且开启了安全组访问控制需要把 DTS 服务的 IP 加白。直接用 SDK 调用查询接口获取当前地域的 DTS 服务 IP 列表from alibabacloud_dts20200101.client import Client from alibabacloud_dts20200101 import models as dts_models client Client(config) req dts_models.DescribeDtsServiceIPRequest( region_idcn-hangzhou-xxx, source_endpoint_instance_typeRDS, destination_endpoint_instance_typeRDS ) resp client.describe_dts_service_ip(req) # 返回的 ip_list 是逗号分隔的字符串按逗号切分后加到白名单 ip_list resp.body.ip_list.split(,) for ip in ip_list: print(f需要放行的 IP: {ip.strip()})这里有个经验一个地域下 DTS 服务 IP 可能有几十个每个 IP 都要放行。如果源库是自建 MySQL还要确认 MySQL 的bind-address允许这些 IP 访问否则即使安全组放行了数据库层依然会拒绝连接。5.2 订阅任务的消费组管理增量数据的第二入口数据订阅是 DTS 里比较容易被低估的能力。通过订阅接口你可以拿到数据库的增量 binlog 数据然后自己写消费端处理实现缓存同步、ES 索引更新等场景。V3.16.0 的开发指南里订阅任务支持消费组的管理接口包括新增订阅任务的消费组CreateConsumerGroup查询订阅任务的消费组详情DescribeConsumerGroup修改消费组密码ModifyConsumerGroup删除消费组DeleteConsumerGroup拿 Python 举例创建一个消费组并查询位点req dts_models.CreateConsumerGroupRequest( subscription_instance_iddtsxxxxxxxxxxxx, consumer_group_namecg-test, consumer_group_user_nameconsumer_user, consumer_group_passwordxxxxxx, region_idcn-hangzhou-xxx ) client.create_consumer_group(req) # 查询消费组详情 query_req dts_models.DescribeConsumerGroupRequest( subscription_instance_iddtsxxxxxxxxxxxx, region_idcn-hangzhou-xxx, page_size10, page_num1 ) resp client.describe_consumer_group(query_req) # consumer_channels 里包含消费组名称、ID、创建时间等信息消费组的用途和 Kafka 的 consumer group 类似多个消费端实例可以共享一个消费组组内分负载组与组之间消费位点互不影响。合理使用消费组可以优雅扩缩容消费者。常见做法是每个业务线单独一个消费组互相隔离消费位点避免一个消费端阻塞影响另一个业务。5.3 把 DTS 任务集成到 CI/CD 流水线项目里最实用的做法是把 DTS 任务的生命周期管理封装成流水线里的一个步骤。比如在测试环境准备阶段自动创建迁移任务并启动在环境销毁阶段自动释放 DTS 实例。关键点是任务启动后要轮询预检查状态和初始化状态而不是启动完就撒手。状态轮询的 Java 逻辑public boolean waitForDtsJobReady(String jobId, int timeoutSeconds) { long deadline System.currentTimeMillis() timeoutSeconds * 1000L; while (System.currentTimeMillis() deadline) { DescribeMigrationJobStatusRequest req new DescribeMigrationJobStatusRequest(); req.setMigrationJobId(jobId); DescribeMigrationJobStatusResponse resp client.describeMigrationJobStatus(req); String status resp.body.getMigrationJobStatus(); // 可能的状态NotStarted、Migrating、Failed、Finished if (Migrating.equals(status) || Finished.equals(status)) { return true; } if (Failed.equals(status)) { throw new RuntimeException(DTS 任务失败请检查预检查详情); } try { Thread.sleep(10000); // 每 10 秒轮询一次 } catch (InterruptedException e) { Thread.currentThread().interrupt(); return false; } } return false; }轮询间隔设 10 秒比较合适太频繁可能把专有云的 API 网关拉进限流状态太短又不够敏感。这里用到的状态字段MigrationJobStatus可能的值在文档的“查看迁移任务状态”一节有完整定义贴到代码注释里方便同事排查。拆完这个文档之后我自己的习惯变了以前拿到专有云 DTS最怕的就是 Endpoint 换个环境就失效每次都要问运维。现在不管在哪个现场我都强制先走一遍完整流程——查版本号、确认是 POP 还是 ASAPI 网关、调用查询 IP 接口、用 CommonRequest 裸调一次 Describe 接口验证网络和鉴权链路全部通了再开始写正式业务代码。这套流程写进项目组的交付 SOP 之后DTS 相关的环境问题从一天缩短到半小时内搞定。专有云的 OpenAPI 本身不难难的是环境信息不透明把这套准备动作固定下来能省掉现场大量来回扯皮的时间希望帮到你。本文还有配套的精品资源点击获取