DAG Generation Engine
Relevant source files
The following files were used as context for generating this wiki page:
The DAG Generation Engine is a core component of the Meltano Airflow extension that automates the creation of Apache Airflow Directed Acyclic Graphs (DAGs) based on Meltano project schedules. It acts as a bridge between Meltano's orchestration metadata and Airflow's execution environment, ensuring that defined data integration pipelines are automatically discoverable and executable by an Airflow scheduler.
The engine operates by querying the Meltano project for its defined schedules and dynamically generating Airflow DAG objects at runtime. This allows users to manage their pipeline schedules within meltano.yml while leveraging Airflow for advanced scheduling, retries, and monitoring.
Sources: airflow_ext/files/orchestrate/meltano.py, README.md:1-15
Architecture and Installation
The engine is delivered as a bundled Python script (meltano.py) that is deployed into the Airflow dags folder during the extension's initialization phase.
Deployment Flow
When a user executes meltano invoke airflow:initialize, the extension performs several setup tasks, including the deployment of the generator.
Key Components
| Component | Description | Source File |
|---|---|---|
Airflow.initialize | Orchestrates the creation of the dags folder and writes the generator script. | airflow_ext/wrapper.py:64 |
meltano.py | The actual generator logic that runs inside the Airflow environment. | airflow_ext/files/orchestrate/meltano.py |
create_dags() | Entry point within meltano.py that calls meltano schedule list. | airflow_ext/files/orchestrate/meltano.py:171 |
BashOperator | The primary Airflow operator used to execute Meltano commands. | airflow_ext/files/orchestrate/meltano.py:105 |
Generation Logic
The engine supports two primary styles of Meltano schedules: legacy "ELT" schedules (v1) and modern "Job" schedules (v2).
Process Flow
The generator follows a structured path to transform Meltano metadata into Airflow objects:
Meltano ELT Generator (v1)
For legacy schedules, the engine creates a singular DAG with a single BashOperator task. The task executes meltano schedule run <schedule_name>.
- DAG ID Pattern:
meltano_<schedule_name> - Task ID:
extract_load - Tags: Includes "meltano", the extractor name, and the loader name.
Sources: airflow_ext/files/orchestrate/meltano.py:53-108
Meltano Job Generator (v2)
For modern job-based schedules, the engine creates a DAG where each task defined in the Meltano job becomes a separate BashOperator task in Airflow.
- DAG ID Pattern:
meltano_<schedule_name>_<job_name> - Task ID Pattern:
<base_id>_task<index> - Dependencies: Tasks are chained linearly;
task(n)is set as a downstream oftask(n-1).
Sources: airflow_ext/files/orchestrate/meltano.py:111-168, tests/test_dag_generator.py:92-98
Configuration and Environment
The engine relies on specific environment variables and directory structures to locate the Meltano binary and project files.
Path Resolution
The generator attempts to find the Meltano executable in the following order:
.meltano/run/binrelative to theMELTANO_PROJECT_ROOT.- Fallback to the system
PATHif the symlink is missing.
Sources: airflow_ext/files/orchestrate/meltano.py:40-50
Default Parameters
All generated DAGs share a common set of default arguments:
| Parameter | Value | Description |
|---|---|---|
catchup | False | Prevents Airflow from backfilling historical intervals. |
max_active_runs | 1 | Ensures only one instance of the pipeline runs at a time. |
retry_delay | 5 minutes | Standard wait time between task retries. |
concurrency | 1 | Limits parallel task execution within the DAG. |
Sources: airflow_ext/files/orchestrate/meltano.py:28-38, airflow_ext/files/orchestrate/meltano.py:98-103
Version Compatibility
The engine includes logic to maintain compatibility across different Airflow versions, specifically handling changes between Airflow 2.x and Airflow 3.x.
Scheduling Syntax
Airflow 3.x introduced changes to the DAG constructor arguments regarding scheduling.
if AIRFLOW_VERSION >= Version("3.0.0"):
dag_kwargs = {"schedule": schedule["interval"]}
else:
dag_kwargs = {"schedule_interval": schedule["interval"]}
Sources: airflow_ext/files/orchestrate/meltano.py:91-95, airflow_ext/files/orchestrate/meltano.py:25
Database Initialization
The Airflow wrapper class detects the version to determine whether to run db init (v2) or db migrate (v3).
Sources: airflow_ext/wrapper.py:159-183
Usage and Customization
Users can interact with the engine via the Meltano CLI. Once initialized, the DAGs are managed by the standard Airflow scheduler.
Initialization Command
meltano invoke airflow:initialize
This command triggers Airflow.initialize(), which creates the configuration and deploys the generator.
Sources: airflow_ext/main.py:24-41, README.md:104-106
Custom DAGs
While the engine automatically generates DAGs for Meltano schedules, users can still define custom DAGs by placing them in the directory defined by AIRFLOW__CORE__DAGS_FOLDER. By default, this is $MELTANO_PROJECT_ROOT/orchestrate/airflow/dags.
Sources: airflow_ext/files/orchestrate/README.md, README.md:41-43
Summary
The DAG Generation Engine provides a hands-off synchronization mechanism between Meltano and Airflow. By dynamically translating Meltano schedules into Airflow DAGs and handling version-specific API differences, it enables a seamless experience for developers using Meltano for ELT and Airflow for orchestration.