Quickstart Guide

Relevant source files

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

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

The airflow-ext utility is a Meltano extension designed to orchestrate Meltano workflows using Apache Airflow. It wraps the Airflow CLI, providing automated configuration generation, database initialization, and a dynamic DAG generator that translates Meltano schedules into Airflow DAGs.

This guide covers the installation, initialization, and basic usage of the extension within a Meltano project. It is built on the Meltano Extension Developer Kit (EDK) and supports Airflow versions 2.7 through 3.x.
Sources: README.md:1-10, pyproject.toml:1-25, airflow_ext/wrapper.py:27-35

Installation and Configuration

To use the extension, it must be added to the utilities section of a meltano.yml file. The extension requires specific environment variables, primarily AIRFLOW_HOME and AIRFLOW_CONFIG, to manage its internal state and configuration files.

Example Meltano Configuration

The following YAML snippet demonstrates a standard configuration for the airflow utility:

  utilities:
  - name: airflow
    namespace: airflow
    pip_url: >
      meltano.edk~=0.6.1
      git+https://github.com/flext-sh/meltano-airflow-ext.git%40main
      apache-airflow==3.2.0
    executable: airflow_invoker
    commands:
      initialize:
        executable: airflow_extension
        args: initialize
      invoke:
        executable: airflow_extension
        args: invoke

Sources: README.md:13-35, pyproject.toml:23-28

Core Environment Variables

The extension relies on environment variables to locate its workspace. If AIRFLOW_HOME is not set, the extension will fail to function.

Variable	Description	Default Path (Example)
AIRFLOW_HOME	Directory for Airflow logs, configs, and DB	$MELTANO_PROJECT_ROOT/orchestrate/airflow
AIRFLOW_CONFIG	Path to the airflow.cfg file	$AIRFLOW_HOME/airflow.cfg
AIRFLOW__CORE__DAGS_FOLDER	Location of Airflow DAG files	$AIRFLOW_HOME/dags
Sources: README.md:78-105, airflow_ext/wrapper.py:46-60

Initialization Process

The initialization command performs three critical setup tasks: generating the default Airflow configuration, initializing the metadata database, and deploying the Meltano DAG generator.

Initialization Flow

The following diagram illustrates the steps taken when meltano invoke airflow:initialize is executed:

The extension uses the Invoker to call the underlying airflow binary. For configuration generation, it captures the output of airflow config list --defaults and strips any leading warning logs before writing to airflow.cfg.
Sources: airflow_ext/wrapper.py:73-108, airflow_ext/wrapper.py:129-168

Usage and Commands

The extension provides two primary entry points:

1. airflow_extension: Handles extension-specific logic like initialize and describe.
2. airflow_invoker: A passthrough shim that allows running any standard Airflow command (e.g., users create, scheduler, webserver).

Common Management Commands

Action	Command
Initialize Environment	meltano invoke airflow:initialize
Create Admin User	meltano invoke airflow users create -u admin -p password --role Admin -e admin@example.com
Start Scheduler	meltano invoke airflow scheduler &
Start Webserver	meltano invoke airflow api-server
Sources: README.md:109-130, airflow_ext/main.py:19-50, airflow_ext/pass_through.py:11-20

Automated DAG Generation

One of the core features of this extension is the meltano_dag_generator.py (deployed from airflow_ext/files/orchestrate/meltano.py). This script dynamically creates Airflow DAGs by querying the Meltano project for schedules.

Data Flow for DAG Discovery

The generator executes a subprocess call to Meltano to retrieve a list of schedules in JSON format and parses them into Airflow objects.

Schedule Types

The generator handles two types of Meltano schedules:

• Legacy ELT (v1): Generates a DAG with a single extract_load task.
• Jobs (v2): Generates a DAG where each Meltano task in the job becomes a sequential BashOperator in the Airflow DAG.
  Sources: airflow_ext/files/orchestrate/meltano.py:55-173

Conclusion

The Meltano Airflow extension simplifies the integration of Airflow as an orchestrator by automating the boilerplate setup of the Airflow environment. By wrapping the CLI and providing dynamic DAG generation, it ensures that Meltano schedules are automatically reflected in the Airflow UI without manual Python DAG development.
Sources: README.md:1-10, airflow_ext/wrapper.py:38-45