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 数据库。
先决条件任务¶
要使用这些操作器,您必须执行以下操作
通过 pip 安装 API 库。
pip install 'apache-airflow[amazon]'详细信息请参阅 Airflow™ 安装
设置连接.
通用参数¶
- 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。此配置可用于配置 避免限制异常、超时等。
{ "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
。
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
。
start_task = DmsStartTaskOperator(
task_id="start_task",
replication_task_arn=task_arn,
)
获取复制任务的详细信息¶
要检索复制任务列表的详细信息,可以使用 DmsDescribeTasksOperator
。
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
。
stop_task = DmsStopTaskOperator(
task_id="stop_task",
replication_task_arn=task_arn,
)
删除复制任务¶
要删除复制任务,可以使用 DmsDeleteTaskOperator
。
delete_task = DmsDeleteTaskOperator(
task_id="delete_task",
replication_task_arn=task_arn,
)
传感器¶
等待复制任务完成¶
要检查复制任务的状态,直到它完成,可以使用 DmsTaskCompletedSensor
。
await_task_stop = DmsTaskCompletedSensor(
task_id="await_task_stop",
replication_task_arn=task_arn,
)