DAG Processor Integration
Relevant source files
The following files were used as context for generating this wiki page:
The DAG Processor Integration in the airflow-ext project serves as a bridge between Meltano's scheduling system and Apache Airflow's orchestration engine. It automates the creation of Airflow Directed Acyclic Graphs (DAGs) by querying the Meltano project's schedules and jobs, then dynamically generating the corresponding Airflow Python files. This integration ensures that data pipelines defined in Meltano are seamlessly recognized and executed by Airflow components like the Scheduler and DAG Processor.
Sources: README.md, airflow_ext/files/orchestrate/meltano.py:1-15
Architecture and Initialization
The integration is managed through an Airflow extension wrapper that handles the initialization of the environment and the deployment of the DAG generation logic. Upon calling meltano invoke airflow:initialize, the extension sets up the necessary directory structures, creates the airflow.cfg file, and installs a bundled script named meltano_dag_generator.py into the Airflow DAGs folder.
Sources: airflow_ext/wrapper.py:61-103, airflow_ext/main.py:24-40
The following diagram illustrates the initialization flow of the integration:
The diagram shows how the extension prepares the Airflow environment and deploys the generator script.
Sources: airflow_ext/wrapper.py:53-118, airflow_ext/main.py:27-33
Dynamic DAG Generation Logic
The core logic resides in airflow_ext/files/orchestrate/meltano.py, which is the template for the generated DAG processor. This script executes meltano schedule list --format=json to retrieve the current state of Meltano's scheduled tasks. It then iterates through these schedules to create Airflow DAG objects.
Sources: airflow_ext/files/orchestrate/meltano.py:175-195
Schedule Types and Processing
The generator handles two primary types of Meltano schedules:
- Legacy ELT Schedules (v1): These are mapped to a single
BashOperatortask that executesmeltano schedule run <name>. - Job-based Schedules (v2): These are mapped to multi-task DAGs where each task in a Meltano job becomes a
BashOperator. Dependencies are maintained between these tasks based on their order in the job definition.
Sources: airflow_ext/files/orchestrate/meltano.py:58-173, tests/test_dag_generator.py:15-56
Data Flow for DAG Discovery
This sequence shows the interaction between the Airflow DAG Processor and the Meltano CLI via the generator script.
Sources: airflow_ext/files/orchestrate/meltano.py:175-199, tests/test_dag_generator.py:59-71
Configuration and Logging
The integration relies on environment variables to locate the Meltano project and configure Airflow's internal paths. These settings ensure that the DAG Processor and Scheduler can locate logs and process files correctly.
Key Environment Settings
| Variable | Description | Default / Example Value |
|---|---|---|
MELTANO_PROJECT_ROOT | Root of the Meltano project | Current working directory |
AIRFLOW__CORE__DAGS_FOLDER | Where Airflow looks for DAG files | $MELTANO_PROJECT_ROOT/orchestrate/airflow/dags |
AIRFLOW__LOGGING__DAG_PROCESSOR_MANAGER_LOG_LOCATION | Path for DAG parser logs | .../logs/dag_processor_manager/dag_processor_manager.log |
AIRFLOW_HOME | Primary Airflow configuration directory | $MELTANO_PROJECT_ROOT/orchestrate/airflow |
Sources: README.md:38-112, airflow_ext/files/orchestrate/meltano.py:38-40
Logic Constraints and Filters
The generator applies several filters during processing:
- Cron Interval: If a schedule interval is set to
@onceor is null, no DAG is generated. - Catchup: The
catchupparameter is explicitly set toFalsebecause Meltano extractors typically do not support date-window extraction. - Airflow Versioning: The generator adjusts the
schedulevsschedule_intervalkeyword based on whether it is running on Airflow 2.x or 3.x.
Sources: airflow_ext/files/orchestrate/meltano.py:65-71, 93-104, 137-142, tests/test_dag_generator.py:100-106
Conclusion
The DAG Processor Integration provides a zero-config-style orchestration experience for Meltano users. By dynamically generating Airflow code from the Meltano project state, it eliminates the need for manual DAG creation while maintaining support for both legacy ELT tasks and modern multi-step job definitions.
Sources: README.md:120-136, airflow_ext/files/orchestrate/meltano.py:195-202