apache-airflow-orchestration

# Apache Airflow Orchestration > Skill by [ara.so](https://ara.so) — Data Skills collection. Apache Airflow is a platform to programmatically author, schedule, and monitor workflows. It allows you to define workflows as Directed Acyclic Graphs (DAGs) in Python code, making them maintainable, versionable, testable, and collaborative. ## Installation ### Using pip ```bash # Install Airflow with constraints for your Python version AIRFLOW_VERSION=3.2.0 PYTHON_VERSION="$(python --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}" ``` ### Using Docker (Recommended for Development) ```bash # Download docker-compose.yaml curl -LfO 'https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml' # Create required directories mkdir -p ./dags ./logs ./plugins ./config # Set the Airflow user echo -e "AIRFLOW_UID=$(id -u)" > .env # Initialize the database docker compose up airflow-init # Start Airflow docker compose up ``` Access the web UI at `http://localhost:8080` (default credentials: `airflow`/`airflow`). ### Standalone Quick Start ```bash # Initialize database and create admin user airflow db init # Create admin user airflow users create \ --username admin \ --firstname Admin \ --lastname User \ --role Admin \ --email admin@example.com # Start the web server (default port 8080) airflow webserver --port 8080 # Start the scheduler (in another terminal) airflow scheduler ``` ## Core Concepts ### DAG (Directed Acyclic Graph) A DAG defines a workflow with tasks and their dependencies. Tasks must not create cycles. ### Basic DAG Structure ```python from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.operators.bash import BashOperator # Default arguments applied to all tasks default_args = { 'owner': 'data-team', 'depends_on_past': False, 'email': ['alerts@example.com'], 'email_on_failure': True, 'email_on_retry': False, 'retries': 3, 'retry_delay': timedelta(minutes=5), } # Define the DAG dag = DAG( 'example_data_pipeline', default_args=default_args, description='A simple data pipeline', schedule='0 0 * * *', # Run daily at midnight (cron expression) start_date=datetime(2024, 1, 1), catchup=False, # Don't run for past dates tags=['example', 'data-engineering'], ) def extract_data(**context): """Extract data from source""" print("Extracting data...") # Your extraction logic here return {'records': 1000} def transform_data(**context): """Transform extracted data""" # Access data from previous task via XCom ti = context['ti'] extracted = ti.xcom_pull(task_ids='extract') print(f"Transforming {extracted['records']} records...") return {'transformed_records': extracted['records']} def load_data(**context): """Load data to destination""" ti = context['ti'] transformed = ti.xcom_pull(task_ids='transform') print(f"Loading {transformed['transformed_records']} records...") # Define tasks extract = PythonOperator( task_id='extract', python_callable=extract_data, dag=dag, ) transform = PythonOperator( task_id='transform', python_callable=transform_data, dag=dag, ) load = PythonOperator( task_id='load', python_callable=load_data, dag=dag, ) # Set task dependencies extract >> transform >> load ``` ### TaskFlow API (Recommended for Airflow 2.0+) Modern, cleaner syntax using decorators: ```python from datetime import datetime from airflow.decorators import dag, task @dag( schedule='@daily', start_date=datetime(2024, 1, 1), catchup=False, tags=['taskflow', 'etl'], ) def taskflow_etl_pipeline(): """ ETL pipeline using TaskFlow API """ @task() def extract(): """Extract data from API""" import requests # Using environment variable for API key import os api_key = os.getenv('DATA_API_KEY') # Simulated extraction data = {'records': [1, 2, 3, 4, 5]} return data @task() def transform(data: dict): """Transform the data""" records = data['records'] transformed = [r * 2 for r in records] return {'transformed': transformed} @task() def load(data: dict): """Load data to warehouse""" print(f"Loading {len(data['transformed'])} records") # Your loading logic here return True # Define pipeline extracted_data = extract() transformed_data = transform(extracted_data) load(transformed_data) # Instantiate the DAG taskflow_etl_pipeline() ``` ## Common Operators ### BashOperator ```python from airflow.operators.bash import BashOperator run_script = BashOperator( task_id='run_data_script', bash_command='python /opt/scripts/process_data.py --date {{ ds }}', dag=dag, ) ``` ### PythonOperator ```python from airflow.operators.python import PythonOperator def my_function(param1, param2, **context): execution_date = context['execution_date'] print(f"Processing for {execution_date}") # Your logic here python_task = PythonOperator( task_id='python_task', python_callable=my_function, op_kwargs={'param1': 'value1', 'param2': 'value2'}, dag=dag, ) ``` ### BranchPythonOperator ```python from airflow.operators.python import BranchPythonOperator from airflow.operators.empty import EmptyOperator def choose_branch(**context): """Decide which branch to execute""" execution_date = context['execution_date'] if execution_date.day % 2 == 0: return 'even_day_task' else: return 'odd_day_task' branch = BranchPythonOperator( task_id='branch_task', python_callable=choose_branch, dag=dag, ) even_task = EmptyOperator(task_id='even_day_task', dag=dag) odd_task = EmptyOperator(task_id='odd_day_task', dag=dag) branch >> [even_task, odd_task] ``` ### EmailOperator ```python from airflow.operators.email import EmailOperator send_email = EmailOperator( task_id='send_notification', to='team@example.com', subject='Pipeline {{ ds }} completed', html_content='<p>The pipeline for {{ ds }} has completed successfully.</p>', dag=dag, ) ``` ## Sensors Sensors wait for a condition to be met before proceeding. ### FileSensor ```python from airflow.sensors.filesystem import FileSensor wait_for_file = FileSensor( task_id='wait_for_data_file', filepath='/data/input/file_{{ ds }}.csv', poke_interval=30, # Check every 30 seconds timeout=3600, # Timeout after 1 hour mode='poke', # 'poke' or 'reschedule' dag=dag, ) ``` ### TimeDeltaSensor ```python from airflow.sensors.time_delta import TimeDeltaSensor from datetime import timedelta wait_sensor = TimeDeltaSensor( task_id='wait_10_minutes', delta=timedelta(minutes=10), dag=dag, ) ``` ### Custom Sensor ```python from airflow.sensors.base import BaseSensorOperator class CustomDataSensor(BaseSensorOperator): def __init__(self, endpoint, **kwargs): super().__init__(**kwargs) self.endpoint = endpoint def poke(self, context): """Check if data is available""" import requests import os api_key = os.getenv('API_KEY') response = requests.get( self.endpoint, headers={'Authorization': f'Bearer {api_key}'} ) return response.status_code == 200 and response.json().get('ready', False) check_data = CustomDataSensor( task_id='check_data_ready', endpoint='https://api.example.com/status', poke_interval=60, timeout=3600, dag=dag, ) ``` ## Connections and Hooks ### Setting Up Connections Connections store credentials and connection details. #### Via CLI ```bash # Add a Postgres connection airflow connections add 'postgres_default' \ --conn-type 'postgres' \ --conn-host 'localhost' \ --conn-schema 'mydb' \ --conn-login 'user' \ --conn-password 'password' \ --conn-port 5432 # Add an HTTP connection airflow connections add 'http_api' \ --conn-type 'http' \ --conn-host 'https://api.example.com' \ --conn-extra '{"api_key": "from_env_var"}' ``` #### Via Environment Variables ```bash export AIRFLOW_CONN_POSTGRES_DEFAULT='postgresql://user:password@localhost:5432/mydb' export AIRFLOW_CONN_HTTP_API='http://api.example.com' ``` ### Using Hooks ```python from airflow.providers.postgres.hooks.postgres import PostgresHook from airflow.decorators import task @task() def query_database(): """Query PostgreSQL database""" hook = PostgresHook(postgres_conn_id='postgres_default') # Execute query and fetch results results = hook.get_records( sql="SELECT * FROM users WHERE created_date = %s", parameters=['2024-01-01'] ) # Or use pandas df = hook.get_pandas_df(sql="SELECT * FROM transactions") return len(results) @task() def insert_data(): """Insert data into database""" hook = PostgresHook(postgres_conn_id='postgres_default') hook.run( sql="INSERT INTO logs (message, timestamp) VALUES (%s, %s)", parameters=[('Pipeline completed', datetime.now())] ) ``` ### HTTP Hook Example ```python from airflow.providers.http.hooks.http import HttpHook from airflow.decorators import task @task() def call_api(): """Make HTTP API call""" hook = HttpHook(http_conn_id='http_api', method='GET') response = hook.run( endpoint='/v1/data', headers={'Content-Type': 'application/json'}, extra_options={'timeout': 30} ) data = response.json() return data ``` ## XCom (Cross-Communication) XCom allows tasks to exchange small amounts of data. ```python from airflow.decorators import dag, task from datetime import datetime @dag(start_date=datetime(2024, 1, 1), schedule=None, catchup=False) def xcom_example(): @task() def push_data(): """Push data to XCom""" return { 'total_records': 1000, 'processing_time': 45.2, 'status': 'success' } @task() def pull_data(data: dict): """Receive data from previous task""" print(f"Received {data['total_records']} records") print(f"Processing took {data['processing_time']} seconds") # Can also use task instance to pull from specific task from airflow.operators.python import get_current_context context = get_current_context() ti = context['ti'] # Pull from specific task specific_data = ti.xcom_pull(task_ids='push_data') return specific_data['status']