Custom Airflow Plugins Integration

Relevant source files

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

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

The Custom Airflow Plugins Integration within airflow-ext provides a managed environment for running Apache Airflow as a Meltano utility. Its primary purpose is to bridge the gap between Meltano's lifecycle (extract, load, transform) and Airflow's orchestration capabilities by automating configuration, database initialization, and DAG generation based on Meltano schedules.

Sources: README.md:1-5, airflow_ext/wrapper.py:24-30

This integration handles the translation of Meltano environment variables into Airflow configurations, ensuring that plugins, DAGs, and logs are stored within the Meltano project structure rather than system-wide locations.

Sources: README.md:23-55

Architecture and Core Components

The integration is built as a Meltano Extension using the Meltano EDK (Extension Development Kit). It wraps the standard Airflow CLI and adds specialized logic for Meltano-specific file placement and initialization.

Component Overview

Component	Responsibility	Source File
Airflow Wrapper	Inherits ExtensionBase to manage AIRFLOW_HOME and configuration files.	airflow_ext/wrapper.py
meltano_dag_generator.py	Dynamically generates Airflow DAGs from Meltano schedules.	airflow_ext/files/orchestrate/meltano.py
Pass-through Invoker	Forwards arbitrary commands directly to the airflow binary.	airflow_ext/pass_through.py
CLI Entrypoint	Typer-based interface for initialize, describe, and invoke.	airflow_ext/main.py

Sources: airflow_ext/wrapper.py:24-40, airflow_ext/main.py:19-58, airflow_ext/pass_through.py:12-21

Data Flow for Initialization

The following diagram illustrates the initialization flow when a user runs meltano invoke airflow:initialize.

The initialization process identifies the Airflow version to determine whether to use db init (Airflow 2.x) or db migrate (Airflow 3.x).

Sources: airflow_ext/wrapper.py:73-100, airflow_ext/wrapper.py:133-162

Configuration Integration

The extension relies on specific environment variables and settings defined in the Meltano project to locate plugins and DAGs.

Key Environment Variables

• AIRFLOW_HOME: The base directory for Airflow files, defaults to $MELTANO_PROJECT_ROOT/orchestrate/airflow.
• AIRFLOW__CORE__PLUGINS_FOLDER: Path where custom Airflow plugins are stored, defaults to $MELTANO_PROJECT_ROOT/orchestrate/airflow/plugins.
• AIRFLOW__CORE__DAGS_FOLDER: Path where DAGs (including the generated Meltano DAGs) are stored.

Sources: README.md:33-40, airflow_ext/wrapper.py:42-55

Configuration Generation Logic

The _create_config method in the Airflow class programmatically generates the airflow.cfg file. It invokes airflow config list --defaults and strips away any warning messages (lines not starting with [ headers) to ensure a valid INI format.

# airflow_ext/wrapper.py:116-126
if proc.stdout:
    config = proc.stdout
    if not config.startswith("["):
        section_start = config.find("\n[")
        if section_start != -1:
            config = config[section_start + 1 :]

Sources: airflow_ext/wrapper.py:108-131

DAG Generation and Plugin Support

The integration automatically installs a DAG generator (meltano_dag_generator.py) that reads Meltano's schedule list and creates corresponding Airflow DAG objects.

DAG Mapping Logic

The generator supports two styles of Meltano schedules:

1. Legacy ELT (v1): Direct extractor-to-loader schedules.
2. Jobs (v2): Complex sequences of tasks defined as Meltano jobs.

Custom DAGs and Plugins

Users can extend the system by placing custom files in the folders defined in meltano.yml.

• Custom DAGs: Placed in $MELTANO_PROJECT_ROOT/orchestrate/airflow/dags.
• Custom Plugins: Placed in $MELTANO_PROJECT_ROOT/orchestrate/airflow/plugins.

Sources: README.md:36-40, airflow_ext/files/orchestrate/README.md:3-5

Command Execution and Pass-through

The extension provides a pass-through mechanism to execute standard Airflow commands (like users create or scheduler) while ensuring the environment is correctly configured for the Meltano project context.

Execution Flow

Summary of Commands

Meltano Command	Extension Action
airflow:initialize	Creates config, initializes DB, and deploys DAG generator.
airflow:invoke [args]	Standard pass-through for CLI commands.
airflow:describe	Returns metadata about the extension.

Sources: airflow_ext/main.py:22-72, airflow_ext/wrapper.py:73-106