Architecture Overview
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 wrap and orchestrate Apache Airflow (versions 2.7+ and 3.x). It serves as a bridge between a Meltano project and an Airflow environment, automating the configuration of Airflow, the initialization of its metadata database, and the generation of Airflow Directed Acyclic Graphs (DAGs) based on Meltano schedules.
Sources: README.md:3-5, pyproject.toml:1-15, airflow_ext/wrapper.py:33-40
Core Components
The architecture consists of three primary layers: the CLI Entry Points, the Extension Wrapper, and the Orchestration Logic.
CLI Entry Points
The project exposes two main command-line interfaces via Python entry points:
airflow_extension: The primary management interface for extension-specific tasks like initialization and description.airflow_invoker: A passthrough shim that allows direct execution of native Airflow commands while maintaining the extension's environmental context.
Sources: pyproject.toml:26-27, airflow_ext/main.py:12-20, airflow_ext/pass_through.py:10-20
Extension Wrapper (Airflow class)
Inheriting from meltano.edk.extension.ExtensionBase, this class manages the Airflow lifecycle. It handles environment variable detection for AIRFLOW_HOME and AIRFLOW_CONFIG, ensures the existence of required directories, and executes pre-invocation tasks.
Sources: airflow_ext/wrapper.py:28-60
Orchestration Logic
A bundled Python script (meltano.py) acts as a dynamic DAG generator. When deployed into the Airflow dags folder, it queries the Meltano project for schedules and programmatically creates corresponding Airflow DAGs.
Sources: airflow_ext/files/orchestrate/meltano.py:1-25
System Flow
The following diagram illustrates the initialization and execution flow when a user interacts with the extension.
The diagram shows how the extension prepares the environment and how the generator dynamically populates Airflow with tasks derived from Meltano.
Sources: airflow_ext/wrapper.py:62-105, airflow_ext/files/orchestrate/meltano.py:205-220
Initialization Logic
The initialization process (triggered by meltano invoke airflow:initialize) ensures the Airflow environment is ready for operation.
- Configuration Generation: The extension runs
airflow config list --defaultsand captures the output to create a localairflow.cfgfile. It strips leading warnings or deprecation notices to ensure a valid.iniformat. - Database Migration: The extension detects the Airflow version. For version 2.x, it executes
db init. For version 3.x, it executesdb migrate. - DAG Generator Deployment: The internal
meltano.pytemplate is written to the configured Airflow DAGs folder asmeltano_dag_generator.py.
Sources: airflow_ext/wrapper.py:107-160
DAG Generation Logic
The meltano_dag_generator.py script dynamically creates DAGs by parsing the output of meltano schedule list --format=json.
Generator Mapping
| Meltano Schedule Type | Airflow Task Type | Description |
|---|---|---|
| Legacy (v1) ELT | BashOperator | Executes meltano schedule run <name>. |
| Job (v2) | Sequential BashOperator | Executes meltano run <task> for each task in the job, setting dependencies sequentially. |
Sources: airflow_ext/files/orchestrate/meltano.py:59-203
Dependency Handling (v2 Jobs)
For Meltano v2 jobs, the generator iterates through the list of tasks defined in the job. It creates a BashOperator for each and uses task.set_upstream(previous_task) to enforce sequential execution within Airflow.
This sequence shows the dynamic registration of DAGs during the Airflow DAG processing cycle.
Sources: airflow_ext/files/orchestrate/meltano.py:205-223
Configuration and Environment
The extension relies heavily on environment variables, typically configured in meltano.yml.
| Setting Name | Environment Variable | Default Path / Value |
|---|---|---|
extension.airflow_home | AIRFLOW_HOME | $MELTANO_PROJECT_ROOT/orchestrate/airflow |
extension.airflow_config | AIRFLOW_CONFIG | $AIRFLOW_HOME/airflow.cfg |
core.dags_folder | AIRFLOW__CORE__DAGS_FOLDER | $AIRFLOW_HOME/dags |
database.sql_alchemy_conn | AIRFLOW__DATABASE__SQL_ALCHEMY_CONN | sqlite:///$MELTANO_PROJECT_ROOT/.../airflow.db |
Sources: README.md:23-60, airflow_ext/wrapper.py:44-55
Summary
airflow-ext acts as a specialized management layer for Apache Airflow within the Meltano ecosystem. By wrapping the Airflow CLI and providing a dynamic DAG generation mechanism, it allows developers to define data integration schedules in Meltano while leveraging Airflow's robust orchestration and monitoring capabilities without manual configuration of the underlying Airflow infrastructure.
Sources: README.md:73-95, airflow_ext/wrapper.py:33-40