Amazon 关系数据库服务 (RDS)¶
Amazon 关系数据库服务 (Amazon RDS) 是一项 Web 服务，可以更轻松地在云中设置、操作和扩展关系数据库。 它为行业标准关系数据库提供经济高效、可调整大小的容量，并管理常见的数据库管理任务。
先决条件任务¶
要使用这些操作器，您需要执行以下操作
通过 pip 安装 API 库。
pip install 'apache-airflow[amazon]'详细信息请参阅 Airflow™ 的安装
设置连接.
操作器¶
创建数据库快照¶
要创建 Amazon RDS 数据库实例或集群的快照，您可以使用 RDSCreateDBSnapshotOperator。 源数据库实例必须处于 available 或 storage-optimization 状态。
create_snapshot = RdsCreateDbSnapshotOperator(
task_id="create_snapshot",
db_type="instance",
db_identifier=rds_instance_name,
db_snapshot_identifier=rds_snapshot_name,
)
复制数据库快照¶
要复制 Amazon RDS 数据库实例或集群的快照，您可以使用 RDSCopyDBSnapshotOperator。 源数据库快照必须处于 available 状态。
copy_snapshot = RdsCopyDbSnapshotOperator(
task_id="copy_snapshot",
db_type="instance",
source_db_snapshot_identifier=rds_snapshot_name,
target_db_snapshot_identifier=rds_snapshot_copy_name,
)
删除数据库快照¶
要删除 Amazon RDS 数据库实例或集群的快照，您可以使用 RDSDeleteDBSnapshotOperator。 数据库快照必须处于 available 状态才能删除。
delete_snapshot = RdsDeleteDbSnapshotOperator(
task_id="delete_snapshot",
db_type="instance",
db_snapshot_identifier=rds_snapshot_name,
)
将 Amazon RDS 快照导出到 Amazon S3¶
要将 Amazon RDS 快照导出到 Amazon S3，您可以使用 RDSStartExportTaskOperator。 提供的 IAM 角色必须有权访问 S3 存储桶。
start_export = RdsStartExportTaskOperator(
task_id="start_export",
export_task_identifier=rds_export_task_id,
source_arn=snapshot_arn,
s3_bucket_name=bucket_name,
s3_prefix="rds-test",
iam_role_arn=test_context[ROLE_ARN_KEY],
kms_key_id=test_context[KMS_KEY_ID_KEY],
)
取消 Amazon RDS 导出任务¶
要取消到 S3 的 Amazon RDS 导出任务，您可以使用 RDSCancelExportTaskOperator。 已经写入 S3 存储桶的任何数据都不会被删除。
cancel_export = RdsCancelExportTaskOperator(
task_id="cancel_export",
export_task_identifier=rds_export_task_id,
)
订阅 Amazon RDS 事件通知¶
要创建 Amazon RDS 事件订阅，您可以使用 RDSCreateEventSubscriptionOperator。 此操作需要 Amazon SNS 主题 Amazon 资源名称 (ARN)。 Amazon RDS 事件通知仅适用于未加密的 SNS 主题。 如果您指定了加密的 SNS 主题，则不会为该主题发送事件通知。
create_subscription = RdsCreateEventSubscriptionOperator(
task_id="create_subscription",
subscription_name=rds_subscription_name,
sns_topic_arn=sns_topic,
source_type="db-instance",
source_ids=[rds_instance_name],
event_categories=["availability"],
)
取消订阅 Amazon RDS 事件通知¶
要删除 Amazon RDS 事件订阅，您可以使用 RDSDeleteEventSubscriptionOperator。
delete_subscription = RdsDeleteEventSubscriptionOperator(
task_id="delete_subscription",
subscription_name=rds_subscription_name,
)
创建数据库实例¶
要创建 AWS 数据库实例，您可以使用 RdsCreateDbInstanceOperator。 您还可以通过将 deferrable 参数设置为 True 以可延迟模式运行此操作器。
create_db_instance = RdsCreateDbInstanceOperator(
task_id="create_db_instance",
db_instance_identifier=rds_db_identifier,
db_instance_class="db.t4g.micro",
engine="postgres",
rds_kwargs={
"MasterUsername": RDS_USERNAME,
"MasterUserPassword": RDS_PASSWORD,
"AllocatedStorage": 20,
"PubliclyAccessible": False,
},
)
删除数据库实例¶
要删除 AWS 数据库实例，您可以使用 RDSDeleteDbInstanceOperator。 您还可以通过将 deferrable 参数设置为 True 以可延迟模式运行此操作器。
delete_db_instance = RdsDeleteDbInstanceOperator(
task_id="delete_db_instance",
db_instance_identifier=rds_db_identifier,
rds_kwargs={
"SkipFinalSnapshot": True,
},
)
启动数据库实例或集群¶
要启动 Amazon RDS 数据库实例或集群，您可以使用 RdsStartDbOperator。
start_db_instance = RdsStartDbOperator(
task_id="start_db_instance",
db_identifier=rds_db_identifier,
)
停止数据库实例或集群¶
要停止 Amazon RDS 数据库实例或集群，您可以使用 RdsStopDbOperator。
stop_db_instance = RdsStopDbOperator(
task_id="stop_db_instance",
db_identifier=rds_db_identifier,
)
传感器¶
等待 Amazon RDS 实例或集群状态¶
要等待 Amazon RDS 实例或集群达到特定状态，您可以使用 RdsDbSensor。 默认情况下，传感器会等待数据库实例达到 available 状态。
await_db_instance = RdsDbSensor(
task_id="await_db_instance",
db_identifier=rds_db_identifier,
)
等待 Amazon RDS 快照状态¶
要等待具有特定状态的 Amazon RDS 快照，您可以使用 RdsSnapshotExistenceSensor。 默认情况下，传感器会等待状态为 available 的快照存在。
snapshot_sensor = RdsSnapshotExistenceSensor(
task_id="snapshot_sensor",
db_type="instance",
db_snapshot_identifier=rds_snapshot_name,
target_statuses=["available"],
)
等待 Amazon RDS 导出任务状态¶
要等待具有特定状态的 Amazon RDS 快照导出任务，您可以使用 RdsExportTaskExistenceSensor。 默认情况下，传感器会等待状态为 available 的快照存在。
export_sensor = RdsExportTaskExistenceSensor(
task_id="export_sensor",
export_task_identifier=rds_export_task_id,
target_statuses=["canceled"],
)