diff --git a/quickstarts/images/dag.png b/quickstarts/images/dag.png index 86a14e20cc..84cc04216c 100644 Binary files a/quickstarts/images/dag.png and b/quickstarts/images/dag.png differ diff --git a/quickstarts/manage-data/airflow-azure-to-teradata-transfer-operator-doc.md b/quickstarts/manage-data/airflow-azure-to-teradata-transfer-operator-doc.md index a1ccde72ee..e68ad54ac2 100644 --- a/quickstarts/manage-data/airflow-azure-to-teradata-transfer-operator-doc.md +++ b/quickstarts/manage-data/airflow-azure-to-teradata-transfer-operator-doc.md @@ -1,220 +1,255 @@ --- sidebar_position: 4 author: Satish Chinthanippu -email: satish.chinthanippu@teradata.com -page_last_update: July 24th, 2024 -description: Transferring CSV, JSON, and Parquet data from Azure Blob Storage to Teradata Vantage with Airflow's Azure Cloud Transfer Operator -keywords: [data warehouses, teradata, vantage, transfer, cloud data platform, object storage, business intelligence, enterprise analytics, airflow, airflow teradata provider, microsoft azure blob storage] +email: satish.chinthanippu@teradata.com, developer.relations@teradata.com +page_last_update: July 22nd, 2026 +description: Transferring CSV, JSON, and Parquet data from Azure Blob Storage to Teradata with Airflow's Azure Cloud Transfer Operator +keywords: [data warehouses, teradata, transfer, cloud data platform, object storage, business intelligence, enterprise analytics, airflow, airflow teradata provider, microsoft azure blob storage] --- import Tabs from '@theme/Tabs'; import TabItem from '@theme/TabItem'; import TrialDocsNote from '../_partials/teradata_trial.mdx' -# Data Transfer from Azure Blob to Teradata Vantage Using Apache Airflow +# Data Transfer from Azure Blob to Teradata Using Apache Airflow ## Overview -This document provides instructions and guidance for transferring data in CSV, JSON and Parquet formats from Microsoft Azure Blob Storage to Teradata Vantage using the Airflow Teradata Provider and the [Azure Cloud Transfer Operator](https://airflow.apache.org/docs/apache-airflow-providers-teradata/stable/operators/azure_blob_to_teradata.html). It outlines the setup, configuration and execution steps required to establish a seamless data transfer pipeline between these platforms. +This document provides instructions and guidance for transferring data in CSV, JSON and Parquet formats from Microsoft Azure Blob Storage to Teradata using the Airflow Teradata Provider and the [Azure Cloud Transfer Operator](https://airflow.apache.org/docs/apache-airflow-providers-teradata/stable/operators/azure_blob_to_teradata.html). It outlines the setup, configuration and execution steps required to establish a seamless data transfer pipeline between these platforms. :::note -Use [The Windows Subsystem for Linux (WSL)](https://learn.microsoft.com/en-us/windows/wsl/install) on `Windows` to try this quickstart example. +On Windows, use [Windows Subsystem for Linux (WSL)](https://learn.microsoft.com/en-us/windows/wsl/install) to complete this quickstart. ::: ## Prerequisites -* Access to a Teradata Vantage instance, version 17.10 or higher. +* Access to a Teradata instance, version 17.10 or higher. -* Python 3.8, 3.9, 3.10 or 3.11 and python3-env, python3-pip installed. - - - ```bash - sudo apt install -y python3-venv python3-pip - ``` - - - ```bash - sudo apt install -y python3-venv python3-pip - ``` - - +* Python 3.10, 3.11, or 3.12 installed. +* `uv` installed. To install `uv` on Linux, WSL, or macOS, run: + + ```bash + curl -LsSf https://astral.sh/uv/install.sh | sh + ``` + + Restart the terminal and verify the installation: + + ```bash + uv --version + ``` + +## Install Apache Airflow + +1. Create and activate a Python virtual environment. + ```bash - brew install python + uv venv airflow_env + source airflow_env/bin/activate ``` - Refer [Installation Guide](https://docs.python-guide.org/starting/install3/osx/) if you face any issues. - - -## Install Apache Airflow -1. Create a new python environment to manage airflow and its dependencies. Activate the environment. +2. Install Apache Airflow 3.2.2. - ``` bash - python3 -m venv airflow_env - source airflow_env/bin/activate - AIRFLOW_VERSION=2.9.3 - PYTHON_VERSION="$(python3 --version | cut -d " " -f 2 | cut -d "." -f 1-2)" - CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt" - pip install "apache-airflow##${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}" + ```bash + AIRFLOW_VERSION=3.2.2 + uv pip install "apache-airflow==${AIRFLOW_VERSION}" ``` -2. Install the Apache Airflow Teradata provider package and the Microsoft Azure provider package. +3. Install the Airflow Teradata provider with Microsoft Azure support. - ``` bash - pip install "apache-airflow-providers-teradata[microsoft.azure]" + ```bash + uv pip install "apache-airflow-providers-teradata[microsoft.azure]" ``` -3. Set the AIRFLOW_HOME environment variable. - ``` bash - export AIRFLOW_HOME=~/airflow +4. Set the `AIRFLOW_HOME` environment variable. + + ```bash + export AIRFLOW_HOME=~/airflow-azure-quickstart ``` ## Configure Apache Airflow -1. Switch to the virtual environment where Apache Airflow was installed at [Install Apache Airflow](./airflow-azure-to-teradata-transfer-operator-doc.md) - ``` bash - source airflow_env/bin/activate - ``` +Configure the environment variables to enable the test connection functionality and prevent sample DAGs from loading in the Airflow UI. -2. Configure the listed environment variables to activate the test connection button, preventing the loading of sample DAGs and default connections in the Airflow UI. +```bash +export AIRFLOW__CORE__TEST_CONNECTION=Enabled +export AIRFLOW__CORE__LOAD_EXAMPLES=false +``` - ``` bash - export AIRFLOW__CORE__TEST_CONNECTION=Enabled - export AIRFLOW__CORE__LOAD_EXAMPLES=false - export AIRFLOW__CORE__LOAD_DEFAULT_CONNECTIONS=false - ``` -## Start the Apache Airflow web server -1. Run airflow's web server +## Start Apache Airflow - ``` bash +1. Start Airflow Standalone. + + ```bash airflow standalone ``` -2. Access the airflow UI. Visit https://localhost:8080 in the browser and log in with the admin account details shown in the terminal. +2. Open [http://localhost:8080](http://localhost:8080) in a browser. Log in with the `admin` username and the password displayed in the terminal. ![Airflow Password](../images/airflow-console-password.png) -## Define the Apache Airflow connection to Vantage +## Define the Apache Airflow connection to Teradata -1. Click on Admin - Connections -2. Click on + to define a new connection to a Teradata Vantage instance. -3. Assign the new connection the id `teradata_default` with Teradata Vantage instance details. - * Connection Id: teradata_default - * Connection Type: Teradata - * Database Server URL (required): Teradata Vantage instance hostname to connect to. - * Database: database name - * Login (required): database user - * Password (required): database user password +1. In the Airflow UI, select **Admin > Connections**. +2. Click **Add Connection**. +3. Configure the connection using the following details: + * Connection Id: `teradata_default` + * Connection Type: `Teradata` + * Database Server URL: Teradata instance hostname. + * Database: Database name. + * Login: Database user. + * Password: Database user password. +4. Save the connection. +5. Open the saved connection and test it. -Refer [Teradata Connection](https://airflow.apache.org/docs/apache-airflow-providers-teradata/stable/connections/teradata.html) for more details. +Refer to the [Teradata Connection](https://airflow.apache.org/docs/apache-airflow-providers-teradata/stable/connections/teradata.html) documentation for more information. -## Define DAG in Apache Airflow -DAGs in airflow are defined as python files. The DAG below transfers data from Teradata-supplied public blob containers to a Teradata Vantage instance. Copy the python code below and save it as `airflow-azure-to-teradata-transfer-operator-demo.py` under the directory $AIRFLOW_HOME/dags. +## Define a DAG in Apache Airflow -This DAG is a very simple example that covers: +Airflow DAGs are defined in Python files. The following DAG transfers CSV data from a Teradata-provided public Azure Blob Storage container to Teradata. -- Droping the destination table if it exists -- Transfer of the data stored in object storage -- Get the number of transferred records -- Write the number of transferred records to the log +1. Create the DAG directory and open a new Python file: -Refer [Azure Blob Storage To Teradata Operator](https://airflow.apache.org/docs/apache-airflow-providers-teradata/stable/_api/airflow/providers/teradata/transfers/azure_blob_to_teradata/index.html) for more information on `Azure Blob Storage to Teradata` Transfer Operator. + ```bash + mkdir -p "$AIRFLOW_HOME/dags" + nano "$AIRFLOW_HOME/dags/airflow-azure-to-teradata-transfer-operator-demo.py" + ``` -``` python +2. Copy the following code into the file: -from __future__ import annotations + ```python + from __future__ import annotations -import datetime + 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 + 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 table if it exists - 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 Vantage instance - 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 from azure blob to teradata table - read_data_table_csv = TeradataOperator( - task_id="read_data_table_csv", - sql="SELECT count(1) from example_blob_teradata_csv;", - ) - # Print number of records in table - 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 - ) + 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: + + ```text + 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](https://airflow.apache.org/docs/apache-airflow-providers-teradata/stable/_api/airflow/providers/teradata/transfers/azure_blob_to_teradata/index.html) documentation for more information. + +## Load the DAG + +After saving the DAG file in the `$AIRFLOW_HOME/dags` directory, Airflow automatically processes it and displays it on the **DAGs** page. The DAG may take a few minutes to appear. + +To check for DAG import errors, open another terminal, activate the virtual environment, set `AIRFLOW_HOME`, and run: + +```bash +source airflow_env/bin/activate +export AIRFLOW_HOME=~/airflow-azure-quickstart +airflow dags list-import-errors ``` -## Load DAG +Refresh the Airflow UI after the DAG has been processed. -When the DAG file is copied to $AIRFLOW_HOME/dags, Apache Airflow displays the DAG in the UI under the DAGs section. It will take 2 to 3 minutes to load the DAG in the Apache Airflow UI. +## Run the DAG -## Run DAG +1. On the **DAGs** page, locate `example_azure_blob_to_teradata_transfer_operator`. +2. Click the **Trigger DAG** icon on the right side of the DAG row. +3. In the dialog box, select **Single Run**, keep the default data interval, and click **Trigger**. +4. Open the DAG to monitor the task execution in the **Grid** view. -Run the dag as shown in the image below. +![Run DAG](../images/dag.png) -![Run dag](../images/dag.png) +## Transfer data from a private Azure Blob Storage container to Teradata -## Transfer data from Private Blob Storage Container to Teradata instance -To successfully transfer data from a Private Blob Storage Container to a Teradata instance, the following prerequisites are necessary. +To transfer data from a private Azure Blob Storage container to Teradata, complete the following prerequisites: -* An Azure account. You can start with a [free account](https://azure.microsoft.com/free/). -* Create an [Azure storage account](https://docs.microsoft.com/en-us/azure/storage/common/storage-quickstart-create-account?tabs=azure-portal) -* Create a [blob container](https://learn.microsoft.com/en-us/azure/storage/blobs/blob-containers-portal) under Azure storage account -* [Upload](https://learn.microsoft.com/en-us/azure/storage/blobs/storage-quickstart-blobs-portal) CSV/JSON/Parquest format files to blob container -* Create a Teradata Authorization object with the Azure Blob Storage Account and the Account Secret Key +* Create an [Azure account](https://azure.microsoft.com/free/). +* Create an [Azure storage account](https://docs.microsoft.com/en-us/azure/storage/common/storage-quickstart-create-account?tabs=azure-portal). +* Create a [blob container](https://learn.microsoft.com/en-us/azure/storage/blobs/blob-containers-portal) in the Azure storage account. +* [Upload](https://learn.microsoft.com/en-us/azure/storage/blobs/storage-quickstart-blobs-portal) CSV, JSON, or Parquet files to the blob container. +* Create a Teradata authorization object using the Azure storage account name and access key. - ``` sql - CREATE AUTHORIZATION azure_authorization USER 'azuretestquickstart' PASSWORD 'AZURE_BLOB_ACCOUNT_SECRET_KEY' + ```sql + CREATE AUTHORIZATION azure_authorization + USER 'azuretestquickstart' + PASSWORD 'AZURE_BLOB_ACCOUNT_ACCESS_KEY'; ``` :::note - Replace `AZURE_BLOB_ACCOUNT_SECRET_KEY` with Azure storage account `azuretestquickstart` [access key](https://learn.microsoft.com/en-us/azure/storage/common/storage-account-keys-manage?toc=%2Fazure%2Fstorage%2Fblobs%2Ftoc.json&bc=%2Fazure%2Fstorage%2Fblobs%2Fbreadcrumb%2Ftoc.json&tabs=azure-portal) + Replace `azuretestquickstart` with the Azure storage account name and `AZURE_BLOB_ACCOUNT_ACCESS_KEY` with its [access key](https://learn.microsoft.com/en-us/azure/storage/common/storage-account-keys-manage?toc=%2Fazure%2Fstorage%2Fblobs%2Ftoc.json&bc=%2Fazure%2Fstorage%2Fblobs%2Fbreadcrumb%2Ftoc.json&tabs=azure-portal). ::: -* Modify `blob_source_key` with YOUR-PRIVATE-OBJECT-STORE-URI in `transfer_data_csv` task and add `teradata_authorization_name` field with Teradata Authorization Object name +* In the `transfer_data_csv` task, replace `YOUR-PRIVATE-OBJECT-STORE-URI` with the URI of the private container and add the `teradata_authorization_name` parameter with the Teradata authorization object name. - ``` python + ```python transfer_data_csv = AzureBlobStorageToTeradataOperator( - task_id="transfer_data_blob_to_teradata_csv", - blob_source_key="YOUR-OBJECT-STORE-URI", - teradata_table="example_blob_teradata_csv", - teradata_conn_id="teradata_default", - teradata_authorization_name="azure_authorization", - trigger_rule="always", - ) + task_id="transfer_data_blob_to_teradata_csv", + blob_source_key="YOUR-PRIVATE-OBJECT-STORE-URI", + teradata_table="example_blob_teradata_csv", + teradata_conn_id="teradata_default", + teradata_authorization_name="azure_authorization", + trigger_rule="always", + ) ``` - ## Summary -This guide details the utilization of the Airflow Teradata Provider’s Azure Cloud Transfer Operator to seamlessly transfer CSV, JSON, and Parquet data from Microsoft Azure Blob Storage to Teradata Vantage, facilitating streamlined data operations between these platforms. + +This guide demonstrated how to use the Airflow Teradata Provider’s Azure Blob Storage to Teradata Transfer Operator to transfer CSV, JSON, and Parquet data from Microsoft Azure Blob Storage to Teradata. ## Further reading -* [Apache Airflow DAGs reference](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dags.html) + +* [Apache Airflow DAGs](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dags.html) * [Teradata Authorization](https://docs.teradata.com/r/Enterprise_IntelliFlex_VMware/SQL-Data-Definition-Language-Syntax-and-Examples/Authorization-Statements-for-External-Routines/CREATE-AUTHORIZATION-and-REPLACE-AUTHORIZATION) -* [Install WSL on windows](https://learn.microsoft.com/en-us/windows/wsl/install) +* [Install WSL on Windows](https://learn.microsoft.com/en-us/windows/wsl/install) \ No newline at end of file