Define a DAG in Apache Airflow

Airflow DAGs are defined in Python files. The following DAG transfers CSV data from a Teradata-provided public Azure Blob Storage container to Teradata.

1. Create the DAG directory and open a new Python file:

   mkdir -p "$AIRFLOW_HOME/dags"
   nano "$AIRFLOW_HOME/dags/airflow-azure-to-teradata-transfer-operator-demo.py"
   

2. Copy the following code into the file:

   from __future__ import annotations
   
   import datetime
   
   from airflow import DAG
   from airflow.operators.bash import BashOperator
   from airflow.providers.teradata.operators.teradata import TeradataOperator
   from airflow.providers.teradata.transfers.azure_blob_to_teradata import AzureBlobStorageToTeradataOperator
   
   DAG_ID = "example_azure_blob_to_teradata_transfer_operator"
   CONN_ID = "teradata_default"
   
   with DAG(
       dag_id=DAG_ID,
       start_date=datetime.datetime(2020, 2, 2),
       schedule="@once",
       catchup=False,
       default_args={"teradata_conn_id": CONN_ID},
   ) as dag:
       # Drop the destination table
       drop_table_if_exists = TeradataOperator(
           task_id="drop_table_if_exists",
           sql="DROP TABLE example_blob_teradata_csv;",
       )
   
       # Transfer data from Azure Blob Storage to Teradata
       transfer_data_csv = AzureBlobStorageToTeradataOperator(
           task_id="transfer_data_blob_to_teradata_csv",
           blob_source_key="/az/akiaxox5jikeotfww4ul.blob.core.windows.net/td-usgs/CSVDATA/09380000/2018/06/",
           public_bucket=True,
           teradata_table="example_blob_teradata_csv",
           teradata_conn_id="teradata_default",
           trigger_rule="always",
       )
   
       # Get the number of records transferred to the Teradata table
       read_data_table_csv = TeradataOperator(
           task_id="read_data_table_csv",
           sql="SELECT COUNT(*) FROM example_blob_teradata_csv;",
       )
   
       # Write the number of records to the task log
       print_number_of_records = BashOperator(
           task_id="print_number_of_records",
           bash_command="echo {{ ti.xcom_pull(task_ids='read_data_table_csv') }}",
       )
   
       (
           drop_table_if_exists
           >> transfer_data_csv
           >> read_data_table_csv
           >> print_number_of_records
       )
   

3. Save the file and exit the editor:

   Ctrl+O
   Enter
   Ctrl+X
   

This DAG performs the following operations:

• Attempts to drop the destination table.
• Transfers data from the public Azure Blob Storage container to Teradata.
• Retrieves the number of transferred records.
• Writes the number of transferred records to the task log.

Refer to the Azure Blob Storage to Teradata Operator documentation for more information.