AWS 数据库迁移服务 (DMS)

AWS 数据库迁移服务 (AWS DMS) 是一种 Web 服务,您可以使用它将数据从本地数据库、Amazon Relational Database Service (Amazon RDS) 数据库实例或 Amazon Elastic Compute Cloud (Amazon EC2) 实例上的数据库迁移到 AWS 服务上的数据库。这些服务可以包括 Amazon RDS 上的数据库或 Amazon EC2 实例上的数据库。您还可以将数据库从 AWS 服务迁移到本地数据库。您可以在使用相同数据库引擎的源端点和目标端点之间进行迁移,例如从 Oracle 数据库迁移到 Oracle 数据库。您还可以在使用不同数据库引擎的源端点和目标端点之间进行迁移,例如从 Oracle 数据库迁移到 PostgreSQL 数据库。

先决条件任务

要使用这些操作器,您必须执行以下操作

通用参数

aws_conn_id

Amazon Web Services 连接 ID 的引用。如果此参数设置为 None,则使用默认的 boto3 行为,不进行连接查找。否则,使用存储在连接中的凭据。默认值:aws_default

region_name

AWS 区域名称。如果此参数设置为 None 或省略,则将使用 AWS 连接额外参数 中的 region_name。否则,使用指定的值而不是连接值。默认值:None

verify

是否验证 SSL 证书。

  • False - 不验证 SSL 证书。

  • path/to/cert/bundle.pem - 要使用的 CA 证书包的文件名。如果要使用与 botocore 使用的 CA 证书包不同的 CA 证书包,可以指定此参数。

如果此参数设置为 None 或省略,则将使用 AWS 连接额外参数 中的 verify。否则,使用指定的值而不是连接值。默认值:None

botocore_config

提供的字典用于构造 botocore.config.Config。此配置可用于配置 避免限制异常、超时等。

示例,有关参数的更多详细信息,请查看 botocore.config.Config
{
    "signature_version": "unsigned",
    "s3": {
        "us_east_1_regional_endpoint": True,
    },
    "retries": {
      "mode": "standard",
      "max_attempts": 10,
    },
    "connect_timeout": 300,
    "read_timeout": 300,
    "tcp_keepalive": True,
}

如果此参数设置为 None 或省略,则将使用 AWS 连接额外参数 中的 config_kwargs。否则,使用指定的值而不是连接值。默认值:None

注意

指定空字典 {} 将覆盖 botocore.config.Config 的连接配置

操作器

创建复制任务

要创建复制任务,可以使用 DmsCreateTaskOperator

tests/system/providers/amazon/aws/example_dms.py[源代码]

create_task = DmsCreateTaskOperator(
    task_id="create_task",
    replication_task_id=dms_replication_task_id,
    source_endpoint_arn=create_assets["source_endpoint_arn"],
    target_endpoint_arn=create_assets["target_endpoint_arn"],
    replication_instance_arn=create_assets["replication_instance_arn"],
    table_mappings=table_mappings,
)

启动复制任务

要启动复制任务,可以使用 DmsStartTaskOperator

tests/system/providers/amazon/aws/example_dms.py[源代码]

start_task = DmsStartTaskOperator(
    task_id="start_task",
    replication_task_arn=task_arn,
)

获取复制任务的详细信息

要检索复制任务列表的详细信息,可以使用 DmsDescribeTasksOperator

tests/system/providers/amazon/aws/example_dms.py[源代码]

describe_tasks = DmsDescribeTasksOperator(
    task_id="describe_tasks",
    describe_tasks_kwargs={
        "Filters": [
            {
                "Name": "replication-instance-arn",
                "Values": [create_assets["replication_instance_arn"]],
            }
        ]
    },
    do_xcom_push=False,
)

停止复制任务

要停止复制任务,可以使用 DmsStopTaskOperator

tests/system/providers/amazon/aws/example_dms.py[源代码]

stop_task = DmsStopTaskOperator(
    task_id="stop_task",
    replication_task_arn=task_arn,
)

删除复制任务

要删除复制任务,可以使用 DmsDeleteTaskOperator

tests/system/providers/amazon/aws/example_dms.py[源代码]

delete_task = DmsDeleteTaskOperator(
    task_id="delete_task",
    replication_task_arn=task_arn,
)

传感器

等待复制任务完成

要检查复制任务的状态,直到它完成,可以使用 DmsTaskCompletedSensor

tests/system/providers/amazon/aws/example_dms.py[源代码]

await_task_stop = DmsTaskCompletedSensor(
    task_id="await_task_stop",
    replication_task_arn=task_arn,
)

此条目是否有帮助?