Apache Airflow

Warnings

• breaking Direct metadata database access from task code is restricted in Airflow 3. Tasks can no longer directly import and use Airflow database sessions or models. All runtime interactions (state transitions, heartbeats, XComs, resource fetching) must now use the dedicated Task Execution API or the official Python API Client.
  Fix: Rewrite task code to use the Task Execution API or the Airflow Python Client for database interactions. Avoid direct SQLAlchemy imports or session usage within task logic. Consider requesting new API endpoints or Task SDK features if required functionality is missing.
• breaking SubDAGs have been removed in Airflow 3. They are replaced by TaskGroups, Assets, and Data Aware Scheduling.
  Fix: Migrate existing SubDAGs to use TaskGroups for grouping related tasks, or explore using Assets and Data Aware Scheduling for more advanced scenarios.
• breaking The Sequential Executor has been removed in Airflow 3. It is replaced by the LocalExecutor, which can still be used with SQLite for local development.
  Fix: Update Airflow configuration to use `LocalExecutor` instead of `SequentialExecutor`.
• deprecated SLAs (Service Level Agreements) are deprecated and have been removed in Airflow 3. They will be replaced by forthcoming Deadline Alerts.
  Fix: Remove SLA definitions from DAGs. Monitor for the introduction of 'Deadline Alerts' as a replacement.
• gotcha Avoid using relative imports in DAG files (e.g., `from . import my_module`). The same DAG file might be parsed in different contexts (scheduler, workers, tests), leading to inconsistent behavior.
  Fix: Always use full Python package paths for imports within DAGs. Ensure shared code is either installed as a Python package or added to `PYTHONPATH` with a unique top-level name to prevent clashes.
• gotcha Do not use Airflow Variables or Connections at the top level of DAG files (i.e., outside of task `execute()` methods or Jinja templates). This can cause slow DAG parsing and unexpected behavior, as the values are fetched every time the DAG file is parsed.
  Fix: Access Airflow Variables and Connections inside operator `execute()` methods, or pass them to operators using Jinja templating, which defers evaluation until task execution. For sensitive data, use Secrets Backend.

Install

• pip install "apache-airflow[celery,cncf.kubernetes,http,postgres,amazon]"==3.1.8 --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-3.1.8/constraints-3.10.txt" Full installation with common extras (PostgreSQL, Celery, Kubernetes, AWS, HTTP) for Python 3.10
• pip install apache-airflow==3.1.8 --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-3.1.8/constraints-3.10.txt" Minimal installation for Python 3.10

Imports

• DAG

  from airflow.models.dag import DAG

• BashOperator

  from airflow.operators.bash import BashOperator

• PythonOperator

  from airflow.operators.python import PythonOperator

• TaskGroup

  from airflow.utils.task_group import TaskGroup

• Provider specific operators (e.g., S3Operator)

  from airflow.providers.amazon.operators.s3 import S3Operator

  Operators specific to external systems are now in provider packages (e.g., airflow.providers.<provider_name>.operators). The old top-level `airflow.operators` path is deprecated for these.
• XComArg

  from airflow.models.xcom_arg import XComArg

  XComArg moved to airflow.models.xcom_arg in Airflow 3.0.

Quickstart

This quickstart defines a simple DAG with Bash and Python operators. To run this locally after installing Airflow, save the code as a `.py` file (e.g., `dags/quickstart_dag.py`) in your `AIRFLOW_HOME/dags` directory. Then, initialize the database and start the Airflow standalone environment. **To set up and run Airflow locally (assuming `apache-airflow` is installed with `sqlite` support via `pip install "apache-airflow[sqlite]"`):** ```bash # (Optional) Set AIRFLOW_HOME, e.g., to a temporary directory export AIRFLOW_HOME=$(pwd)/airflow_home # Initialize the database and create an admin user (first time only) airflow standalone # Follow prompts to set admin password. This command also starts webserver, scheduler, and triggerer. # You can also start components separately: # airflow db migrate # airflow users create --username admin --firstname Airflow --lastname Admin --role Admin --email admin@example.com -p mypassword # airflow webserver --port 8080 # airflow scheduler # airflow triggerer # After starting, visit http://localhost:8080 to enable the DAG. ``` Remember to define `AIRFLOW_HOME` before running `airflow standalone` or `airflow db init`.

import os
from datetime import datetime

from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator


# Set AIRFLOW_HOME if not already set (e.g., in a local dev setup)
# os.environ['AIRFLOW_HOME'] = os.environ.get('AIRFLOW_HOME', '~/airflow')


def _greet(name):
    print(f"Hello, {name} from a Python task!")


with DAG(
    dag_id='simple_airflow_quickstart',
    start_date=datetime(2023, 1, 1),
    schedule_interval='@daily',
    catchup=False,
    tags=['quickstart'],
) as dag:
    start_task = BashOperator(
        task_id='start_workflow',
        bash_command='echo "Starting the workflow!"',
    )

    greet_task = PythonOperator(
        task_id='greet_with_python',
        python_callable=_greet,
        op_kwargs={'name': 'Airflow User'},
    )

    end_task = BashOperator(
        task_id='end_workflow',
        bash_command='echo "Workflow finished!"',
    )

    start_task >> greet_task >> end_task