AWS Database Migration Service Operators¶

Prerequisite Tasks¶

To use these operators, you must do a few things:

Create necessary resources using AWS Console or AWS CLI.
Install API libraries via pip.
pip install 'apache-airflow[amazon]'
Detailed information is available Installation
Setup Connection.

Overview¶

Airflow to AWS Database Migration Service (DMS) integration provides several operators to create and interact with DMS replication tasks.

DmsTaskBaseSensor

DmsTaskCompletedSensor

DmsCreateTaskOperator

DmsDeleteTaskOperator

DmsDescribeTasksOperator

DmsStartTaskOperator

DmsStopTaskOperator

One example_dag is provided which showcases some of these operators in action.

example_dms_full_load_task.py

Create replication task, wait for it completion and delete it.¶

Purpose¶

This example DAG example_dms_full_load_task.py uses DmsCreateTaskOperator, DmsStartTaskOperator, DmsTaskCompletedSensor, DmsDeleteTaskOperator to create replication task, start it, wait for it to be completed, and then delete it.

Defining tasks¶

In the following code we create a new replication task, start it, wait for it to be completed and then delete it.

airflow/providers/amazon/aws/example_dags/example_dms_full_load_task.py[source]

    create_task = DmsCreateTaskOperator(
        task_id='create_task',
        replication_task_id=REPLICATION_TASK_ID,
        source_endpoint_arn=SOURCE_ENDPOINT_ARN,
        target_endpoint_arn=TARGET_ENDPOINT_ARN,
        replication_instance_arn=REPLICATION_INSTANCE_ARN,
        table_mappings=TABLE_MAPPINGS,
    )

    start_task = DmsStartTaskOperator(
        task_id='start_task',
        replication_task_arn=create_task.output,
    )

    wait_for_completion = DmsTaskCompletedSensor(
        task_id='wait_for_completion',
        replication_task_arn=create_task.output,
    )

    delete_task = DmsDeleteTaskOperator(
        task_id='delete_task',
        replication_task_arn=create_task.output,
    )

Reference¶

For further information, look at:

Boto3 Library Documentation for DMS