DAG Deployment and Orchestration
Relevant source files
The following files were used as context for generating this wiki page:
The DAG Deployment and Orchestration system in airflow-ext provides a seamless bridge between Meltano scheduled tasks and Apache Airflow. It automates the discovery of Meltano schedules and dynamically generates Airflow Directed Acyclic Graphs (DAGs) to execute them. This allows users to manage their data pipelines in Meltano while leveraging Airflow's robust scheduling and monitoring capabilities.
The core of this system is an "orchestrator" generator that is automatically deployed during the extension's initialization phase. This generator queries the Meltano project for existing schedules and translates them into native Airflow objects, ensuring that data extraction, loading, and transformation (ELT) jobs are executed according to their defined cron intervals.
Sources: README.md:1-10, airflow_ext/wrapper.py:72-88, airflow_ext/files/orchestrate/meltano.py:1-20
Initialization and Deployment Flow
The deployment process begins with the initialize command. This command sets up the necessary environment for Airflow to function within a Meltano project, including configuration generation and the installation of the DAG generator script.
The diagram above illustrates the sequential steps taken by the extension to prepare the Airflow environment and deploy the dynamic DAG generation logic.
Sources: airflow_ext/wrapper.py:65-98, airflow_ext/main.py:27-44
Core Initialization Tasks
| Task | Description | Implementation Detail |
|---|---|---|
| Configuration | Generates a default airflow.cfg by querying Airflow defaults. | _create_config() in wrapper.py |
| Database Setup | Runs db init (Airflow 2) or db migrate (Airflow 3). | _initdb() in wrapper.py |
| Generator Deployment | Copies the internal meltano.py template to the Airflow DAGs folder. | initialize() in wrapper.py |
Sources: airflow_ext/wrapper.py:100-155
Dynamic DAG Generation Logic
Once deployed, the meltano_dag_generator.py script runs within the Airflow scheduler's context. It executes a subprocess call to meltano schedule list --format=json to retrieve all defined schedules.
Orchestration Components
The generator handles two primary types of schedules defined in Meltano:
- Legacy ELT Schedules (v1): Singular tasks that run a specific extractor and loader pair.
- Job-based Schedules (v2): Complex sequences of tasks defined as a named "job" in Meltano.
The sequence diagram shows how the Airflow scheduler interacts with the generator script to dynamically discover and register DAGs based on the current Meltano project state.
Sources: airflow_ext/files/orchestrate/meltano.py:168-190, tests/test_dag_generator.py:70-100
DAG Mapping Logic
- ID Convention: DAGs are prefixed with
meltano_followed by the schedule name. For job-based schedules, the format ismeltano_{schedule_name}_{job_name}. - Execution: All tasks use the
BashOperatorto execute Meltano commands (e.g.,meltano schedule run <name>ormeltano run <task_args>). - Dependencies: For job-based schedules, the generator preserves the order of tasks by setting up-stream dependencies sequentially.
- Intervals: The generator respects the
cron_intervaldefined in Meltano. Schedules with@onceor no interval are explicitly skipped to prevent unnecessary DAG creation.
Sources: airflow_ext/files/orchestrate/meltano.py:53-125, airflow_ext/files/orchestrate/meltano.py:128-165
Configuration and Environment
The orchestration system relies on specific environment variables and directory structures to locate the Meltano binary and the project root.
| Environment Variable | Description | Default / Source |
|---|---|---|
MELTANO_PROJECT_ROOT | Path to the Meltano project. | Current working directory if unset |
AIRFLOW_HOME | Directory for Airflow configs and logs. | $MELTANO_PROJECT_ROOT/orchestrate/airflow |
AIRFLOW__CORE__DAGS_FOLDER | Location where DAGs are stored and discovered. | $AIRFLOW_HOME/dags |
MELTANO_BIN | Path to the meltano executable. | .meltano/run/bin or meltano in PATH |
Sources: README.md:25-60, airflow_ext/files/orchestrate/meltano.py:38-50, airflow_ext/wrapper.py:37-55
Manual DAG Definition
Users can define custom Airflow DAGs that are not derived from Meltano schedules. Any .py file placed in the AIRFLOW__CORE__DAGS_FOLDER (defaulting to $MELTANO_PROJECT_ROOT/orchestrate/airflow/dags) will be automatically picked up by the Airflow scheduler alongside the generated Meltano DAGs.
Sources: airflow_ext/files/orchestrate/README.md:1-5
Summary
DAG Deployment and Orchestration in this extension provides an automated, low-configuration method to run Meltano pipelines. By deploying a dynamic generator that translates Meltano's internal scheduling metadata into Airflow's native DAG format, the system ensures that changes in meltano.yml are automatically reflected in the Airflow UI without manual intervention. This architecture supports both legacy ELT patterns and modern task-based jobs while maintaining compatibility across Airflow 2.x and 3.x versions.