AWS是否提供API端点获取DMS数据库迁移任务的mapping rules？

需求背景 #

• 需要以编程方式检索数据库迁移任务的mapping rules，判断复制任务是否在源表上设置了filter-type规则
• 从PostgreSQL读取数据且在非主键列上应用源过滤器时，DMS迁移任务可能出现问题

已尝试操作 #

• 在AWS控制台的AWS DMS -> Database Migration Tasks -> <TaskName>下能看到Mapping rules标签页
• 调用Boto3的describe_endpoint和describe_replication_tasks接口，但未找到rules相关内容

期望结果 #

能否通过编程方式获取如下类似的JSON格式数据？若可以，具体方式是什么？

{
    "rules": [{
        "rule-type": "selection",
        "rule-id": "1",
        "rule-name": "1",
        "object-locator": {
            "schema-name": "test",
            "table-name": "employee"
        },
        "rule-action": "include",
        "filters": [{
            "filter-type": "source",
            "column-name": "empid",
            "filter-conditions": [{
                "filter-operator": "gte",
                "value": "100"
            }]
        }]
    }]
} 

可以通过编程方式获取DMS迁移任务的mapping rules，具体操作如下：

核心思路 #

调用describe_replication_tasks接口后，从返回结果的ReplicationTaskSettings字段中解析出嵌套的TableMappings配置，其中就包含你需要的rules数组。

Boto3示例代码 #

import boto3
import json

# 初始化DMS客户端
dms_client = boto3.client('dms')

# 指定目标迁移任务的ID或ARN
task_response = dms_client.describe_replication_tasks(
    Filters=[
        {
            'Name': 'replication-task-id',
            'Values': ['你的迁移任务ID/ARN']
        }
    ]
)

# 提取并解析任务配置
task_settings_str = task_response['ReplicationTasks'][0]['ReplicationTaskSettings']
task_settings = json.loads(task_settings_str)

# 解析TableMappings获取rules
table_mappings_str = task_settings['TableMappings']
table_mappings = json.loads(table_mappings_str)
target_rules = table_mappings['rules']

# 输出结果或进行后续检查
print(json.dumps(target_rules, indent=2))

关键细节 #

• ReplicationTaskSettings是JSON格式的字符串，需要先反序列化为Python字典，才能提取TableMappings字段
• TableMappings同样是JSON字符串，需再次反序列化才能拿到rules数组
• 遍历rules数组即可检查每个规则的filters字段，判断是否存在filter-type为source的规则，以及对应的列属性

内容的提问来源于stack exchange，提问作者J Weezy