Meltano Configuration Definition
Relevant source files
The following files were used as context for generating this wiki page:
The Meltano Configuration Definition for the Airflow extension facilitates the seamless integration of Apache Airflow as an orchestrator within a Meltano project. It defines how Airflow environments are provisioned, how its internal configuration files (such as airflow.cfg) are generated from Meltano settings, and how the metadata database is initialized or migrated based on the installed Airflow version.
The configuration system bridges Meltano's environment-variable-driven settings with Airflow's native configuration requirements. It ensures that critical paths for Directed Acyclic Graphs (DAGs), logs, and plugins are correctly mapped to the Meltano project structure.
Sources: README.md:12-88, airflow_ext/wrapper.py:33-54
Core Configuration Components
The extension utilizes a combination of environment variables and structured settings defined in the meltano.yml file to manage the Airflow instance.
Environment Variable Mapping
The extension maps Meltano settings directly to Airflow environment variables using the env: key in the utility definition. This allows Meltano to inject project-specific paths into the Airflow process.
| Meltano Setting Name | Airflow Environment Variable | Default/Example Value |
|---|---|---|
database.sql_alchemy_conn | AIRFLOW__DATABASE__SQL_ALCHEMY_CONN | sqlite:///$MELTANO_PROJECT_ROOT/.meltano/utilities/airflow/airflow.db |
core.dags_folder | AIRFLOW__CORE__DAGS_FOLDER | $MELTANO_PROJECT_ROOT/orchestrate/airflow/dags |
extension.airflow_home | AIRFLOW_HOME | $MELTANO_PROJECT_ROOT/orchestrate/airflow |
extension.airflow_config | AIRFLOW_CONFIG | $MELTANO_PROJECT_ROOT/orchestrate/airflow/airflow.cfg |
Sources: README.md:31-88, airflow_ext/wrapper.py:41-52
Configuration Generation Logic
When the extension is initialized, it dynamically generates the airflow.cfg file. This is achieved by invoking the underlying Airflow binary to retrieve default settings and then capturing that output into the path specified by AIRFLOW_CONFIG.
The logic specifically handles cases where Airflow might emit deprecation warnings or informational text before the actual INI content by searching for the first occurrence of a section header (e.g., [).
Sources: airflow_ext/wrapper.py:117-142.%0ASources%3A%20%5Bairflow_ext/wrapper.py%3A117-142)
Initialization and Database Management
The configuration definition includes logic to handle differences between Airflow versions, specifically between version 2.x and 3.x, regarding database setup.
Database Command Selection
The extension detects the Airflow version to determine which command to run for metadata storage setup:
- Airflow 2.x: Executes
airflow db init. - Airflow 3.x: Executes
airflow db migrate.
Sources: airflow_ext/wrapper.py:144-170
DAG Generation Configuration
A key part of the Meltano configuration is the meltano_dag_generator.py, which is automatically deployed to the AIRFLOW__CORE__DAGS_FOLDER during the initialize command.
Meltano Schedule Integration
The generator identifies schedules within the Meltano project by executing meltano schedule list --format=json. It configures DAGs differently based on whether they are legacy "ELT" schedules or modern "Job" schedules.
| Feature | Legacy ELT Schedule | Job-based Schedule |
|---|---|---|
| Command | meltano schedule run <name> | meltano run <task> |
| Task Count | Single extract_load task | Multiple tasks based on job definition |
| ID Format | meltano_<schedule_name> | meltano_<schedule_name>_<job_name> |
Sources: airflow_ext/files/orchestrate/meltano.py:53-155, airflow_ext/wrapper.py:75-87
Default DAG Arguments
The generated DAGs are configured with a standard set of default arguments to ensure reliable execution within the Meltano context:
- Catchup: Set to
Falseby default to avoid enqueuing unnecessary historical jobs. - Concurrency: Defaulted to
1. - Retries: Defaulted to
1with a 5-minute delay.
Sources: [airflow_ext/files/orchestrate/meltano.py:31-41], airflow_ext/files/orchestrate/meltano.py:88-90
Path Resolution and Environment
The configuration relies heavily on the MELTANO_PROJECT_ROOT environment variable. The extension uses os.path.expandvars to ensure that paths defined in settings (like the DAGs folder) are correctly resolved to absolute paths on the file system.
self.airflow_core_dags_path = Path(
os.path.expandvars(
os.environ.get(
"AIRFLOW__CORE__DAGS_FOLDER",
f"{self.airflow_home}/dags",
)
)
)
Sources: airflow_ext/wrapper.py:46-54, airflow_ext/files/orchestrate/meltano.py:44
Summary
The Meltano Configuration Definition provides a structured way to manage Airflow settings through Meltano's unified interface. By automating the creation of airflow.cfg, managing database migrations across versions, and deploying a dynamic DAG generator that reads Meltano schedules, the extension ensures that Airflow remains synchronized with the Meltano project state.