Introduction
Relevant source files
The following files were used as context for generating this wiki page:
The airflow-ext project is a Meltano utility extension designed to integrate Apache Airflow as an orchestrator within a Meltano project. It leverages the Meltano Extension SDK (meltano.edk) to wrap the Airflow CLI, providing automated configuration, database initialization, and a dynamic DAG generator that translates Meltano schedules into Airflow DAGs.
Sources: pyproject.toml:1-25, README.md:1-5, airflow_ext/wrapper.py:23-30
This extension acts as a bridge, ensuring that Airflow environment variables, configuration files (airflow.cfg), and metadata databases are correctly initialized within the Meltano project structure. It supports both Airflow 2.7+ and Airflow 3.x, handling version-specific differences in database migrations and command syntax.
Sources: README.md:7-10, airflow_ext/wrapper.py:127-148
Core Architecture
The extension is built on three primary components: the Airflow wrapper class, a Typer-based CLI for extension commands, and a pass-through invoker for direct Airflow CLI access.
Component Relationship
The following diagram illustrates how the Meltano extension interacts with the underlying Airflow binary and the Meltano project environment.
The Airflow Wrapper
The Airflow class, inheriting from ExtensionBase, manages the lifecycle of the Airflow instance. It identifies the AIRFLOW_HOME and AIRFLOW_CONFIG paths from the environment and ensures the directory structure exists before executing commands.
Sources: airflow_ext/wrapper.py:28-56
Initialization Logic
The initialize command is a critical feature that automates the setup of the Airflow environment. When executed, it performs the following steps:
- Configuration Generation: Runs
airflow config list --defaultsto generate a baseairflow.cfgfile, cleaning up any non-INI stdout data such as deprecation warnings. - Database Migration: Detects the Airflow version. It runs
db initfor Airflow 2.x anddb migratefor Airflow 3.x. - DAG Generator Deployment: Copies the bundled
meltano.pyDAG generator into the project's DAGs folder.
Sources: airflow_ext/wrapper.py:65-103, airflow_ext/wrapper.py:127-148
Initialization Sequence
Meltano DAG Generator
The extension includes a dynamic DAG generator (meltano.py) that is installed into the Airflow DAGs folder. This script queries the Meltano project for schedules and job definitions to create corresponding Airflow DAGs.
Schedule Translation
- ELT Schedules (v1): Generates a DAG with a single
BashOperatortask that runsmeltano schedule run <name>. - Job Schedules (v2): Generates a DAG where each task in the Meltano job is mapped to a sequential
BashOperatortask, preserving execution order. - Exclusions: Schedules with an interval of
@once(wherecron_intervalis null) are skipped to prevent unnecessary DAG creation.
Sources: airflow_ext/files/orchestrate/meltano.py:59-166
Support for Airflow Versions
The generator dynamically adjusts to the installed Airflow version for task imports and DAG arguments:
| Feature | Airflow 2.x | Airflow 3.x |
|---|---|---|
| BashOperator Import | airflow.operators.bash | airflow.providers.standard.operators.bash |
| Scheduling Arg | schedule_interval | schedule |
Sources: airflow_ext/files/orchestrate/meltano.py:25-30, airflow_ext/files/orchestrate/meltano.py:97-100, airflow_ext/files/orchestrate/meltano.py:133-136
Configuration and Settings
The extension relies on several environment variables typically configured in the meltano.yml file.
| Setting | Environment Variable | Default Path / Value |
|---|---|---|
| Airflow Home | AIRFLOW_HOME | $MELTANO_PROJECT_ROOT/orchestrate/airflow |
| Config File | AIRFLOW_CONFIG | $AIRFLOW_HOME/airflow.cfg |
| DAGs Folder | AIRFLOW__CORE__DAGS_FOLDER | $AIRFLOW_HOME/dags |
| SQL Connection | AIRFLOW__DATABASE__SQL_ALCHEMY_CONN | sqlite:///$MELTANO_PROJECT_ROOT/.meltano/utilities/airflow/airflow.db |
Sources: README.md:27-55, airflow_ext/wrapper.py:44-54
CLI Commands
The extension provides two entry points defined in pyproject.toml: airflow_extension for management tasks and airflow_invoker for direct passthrough.
initialize: Prepares the environment (config, DB, DAGs).describe: Returns the extension metadata and available commands.invoke: Allows calling the extension logic via Typer.airflow_invoker: A passthrough shim that forwards all arguments directly to the Airflow CLI while maintaining Meltano's logging context.
Sources: pyproject.toml:34-36, airflow_ext/main.py:28-65, airflow_ext/pass_through.py:14-22
Summary
The airflow-ext utility provides a managed environment for running Apache Airflow within a Meltano project. By automating the boilerplate of configuration generation and database migrations, and providing a dynamic DAG generator, it allows developers to focus on orchestration logic while ensuring compatibility across different Airflow versions.