Production Deployment Patterns

Relevant source files

The following files were used as context for generating this wiki page:

• README.md
• airflow_ext/wrapper.py
• airflow_ext/files/orchestrate/meltano.py
• airflow_ext/main.py
• pyproject.toml
• airflow_ext/pass_through.py

The airflow-ext utility serves as a Meltano Airflow extension, designed to facilitate the deployment and orchestration of Meltano projects using Apache Airflow. It provides a structured way to initialize Airflow environments, generate DAGs automatically based on Meltano schedules, and manage Airflow service components within a Meltano-managed lifecycle.

The extension supports both Airflow 2.7+ and Airflow 3.x, handling version-specific differences such as database initialization commands (db init vs db migrate) and scheduling parameters (schedule_interval vs schedule).

Sources: README.md:1-7, airflow_ext/wrapper.py:165-177, airflow_ext/files/orchestrate/meltano.py:90-94

Extension Architecture and Initialization

The extension is built on the Meltano Extension Developer Kit (EDK) and wraps the standard Airflow CLI. The Airflow class in wrapper.py implements the core logic for configuration and environment setup.

Initialization Flow

When meltano invoke airflow:initialize is called, the extension performs several critical setup tasks:

1. Config Generation: It generates an initial airflow.cfg by capturing the default output of the airflow config list --defaults command.
2. Database Migration: It detects the Airflow version and runs the appropriate database setup command.
3. DAG Generation Deployment: It auto-generates a meltano_dag_generator.py file into the Airflow DAGs folder. This script is responsible for dynamically creating Airflow DAGs from the meltano schedule list.

Automated DAG Generation

The extension includes a dynamic DAG generator (meltano.py) that translates Meltano schedules into Airflow DAGs. It supports two primary schedule formats:

1. Legacy ELT Schedules

For older Meltano schedules, the generator creates a DAG containing a single BashOperator task named extract_load. This task executes meltano schedule run <schedule_name>.

2. Job-based Schedules (Meltano v2)

For modern Meltano job schedules, the generator creates a sequence of BashOperator tasks. Each task in the Meltano job is represented as an Airflow task, with upstream dependencies automatically configured to ensure sequential execution.

Production Service Management

In a production environment, various Airflow components must be started as persistent processes. The extension allows these to be invoked via meltano invoke.

Service	Command	Backgrounding	Description
Scheduler	meltano invoke airflow scheduler	Recommended (&)	Monitors DAGs and triggers task instances.
Triggerer	meltano invoke airflow triggerer	Recommended (&)	Runs event loops for deferrable operators.
DAG Processor	meltano invoke airflow dag-processor	Recommended (&)	Parses DAG files to keep them updated in the DB.
Web / API Server	meltano invoke airflow api-server	Foreground	Provides the UI and API access.

Sources: README.md:104-114

Configuration and Environment Variables

The extension relies heavily on environment variables to define the Airflow home directory and connection strings. These are typically mapped in the meltano.yml file.

Meltano Setting	Environment Variable	Default Path / Value
extension.airflow_home	AIRFLOW_HOME	$MELTANO_PROJECT_ROOT/orchestrate/airflow
extension.airflow_config	AIRFLOW_CONFIG	$AIRFLOW_HOME/airflow.cfg
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
logging.base_log_folder	AIRFLOW__LOGGING__BASE_LOG_FOLDER	$MELTANO_PROJECT_ROOT/.meltano/utilities/airflow/logs

Sources: README.md:33-91, airflow_ext/wrapper.py:38-51

Implementation Details: Passthrough Invoker

The extension provides a pass_through_invoker to allow any standard Airflow CLI command to be executed within the Meltano-configured environment. This is handled by airflow_ext/pass_through.py, which shims the CLI calls.

def pass_through_cli() -> None:
    """Pass through CLI entry point."""
    pass_through_logging_config()
    ext = Airflow()
    ext.pass_through_invoker(
        structlog.getLogger("airflow_invoker"),
        *sys.argv[1:] if len(sys.argv) > 1 else [],
    )

Sources: airflow_ext/pass_through.py:10-18, pyproject.toml:34

Summary

Production deployment of airflow-ext involves initializing the environment to generate configuration and the dynamic DAG generator, followed by running the Airflow scheduler, webserver, and other components as background services. By using the bundled meltano.py generator, the system ensures that any pipeline defined in Meltano is automatically reflected as an Airflow DAG without manual operator definition.