Scheduler Service Management
Relevant source files
The following files were used as context for generating this wiki page:
Scheduler Service Management in the airflow-ext project involves the orchestration of Apache Airflow components to manage Meltano-based data integration pipelines. This extension acts as a bridge, allowing Meltano to leverage Airflow as a scheduler and orchestrator by automatically translating Meltano schedules into Airflow Directed Acyclic Graphs (DAGs).
The system manages the lifecycle of the Airflow scheduler and its associated services (webserver, triggerer, dag-processor), ensuring that Meltano schedules—defined as legacy ELT tasks or modern Job-based tasks—are executed reliably according to their defined cron intervals.
Sources: README.md:1-5, airflow_ext/files/orchestrate/meltano.py:12-25
Service Architecture and Orchestration
The orchestration architecture relies on a specialized extension wrapper that manages the initialization and invocation of Airflow services within a Meltano project context.
Component Interaction Flow
The following diagram illustrates how the airflow-ext initialization process sets up the environment for service management.
The extension handles differences between Airflow 2.x and 3.x, such as using db init versus db migrate and adjusting the API/Webserver port configuration.
Sources: airflow_ext/wrapper.py:144-167, README.md:104-124, airflow_ext/main.py:27-44
Core Service Commands
The scheduler and its supporting services are managed via the Meltano CLI through passthrough commands.
| Service | Execution Command | Description |
|---|---|---|
| Scheduler | meltano invoke airflow scheduler | The main orchestration engine that monitors tasks and DAGs. |
| API/Web Server | meltano invoke airflow api-server | Provides the UI and API access (Airflow 3.x uses api-server). |
| Triggerer | meltano invoke airflow triggerer | Handles deferred tasks/async operators. |
| DAG Processor | meltano invoke airflow dag-processor | Separated process for parsing DAG files. |
Sources: README.md:118-124, airflow_ext/wrapper.py:33-35
DAG Generation Logic
The heart of the Scheduler Service Management is the meltano_dag_generator.py (sourced from airflow_ext/files/orchestrate/meltano.py). It dynamically constructs DAGs by querying the Meltano API for existing schedules.
Translation Process
- Discovery: The generator executes
meltano schedule list --format=jsonto retrieve project schedules. - Filtering: Schedules with
@onceintervals or missing cron intervals are ignored. - Construction:
- Legacy ELT: Creates a single
BashOperatortask callingmeltano schedule run <name>. - Meltano Jobs: Creates a sequence of
BashOperatortasks for each task in the Meltano job, maintaining upstream dependencies.
- Legacy ELT: Creates a single
Task Implementation
Each task within an Airflow DAG is implemented as a BashOperator. The operator executes commands by navigating to the MELTANO_PROJECT_ROOT and invoking the meltano binary found in .meltano/run/bin or the system PATH.
Sources: airflow_ext/files/orchestrate/meltano.py:39-50, airflow_ext/files/orchestrate/meltano.py:100-104
Configuration and Environment
Service management is heavily dependent on environment variables that define paths for logs, databases, and DAG storage.
Key Configuration Settings
| Setting | Default Value (relative to project root) | Airflow Env Var |
|---|---|---|
| SQL Alchemy Conn | .meltano/utilities/airflow/airflow.db | AIRFLOW__DATABASE__SQL_ALCHEMY_CONN |
| DAGs Folder | orchestrate/airflow/dags | AIRFLOW__CORE__DAGS_FOLDER |
| Airflow Home | orchestrate/airflow | AIRFLOW_HOME |
| Scheduler Logs | .meltano/utilities/airflow/logs/scheduler | AIRFLOW__SCHEDULER__CHILD_PROCESS_LOG_DIRECTORY |
Sources: README.md:32-96, airflow_ext/wrapper.py:37-52
Environment Initialization
The Airflow wrapper class (in airflow_ext/wrapper.py) ensures that AIRFLOW_HOME is present and establishes the path to airflow.cfg. During the initialize command, it captures the default Airflow configuration by running airflow config list --defaults and stripping any leading warnings to ensure a valid .ini file is written.
Sources: airflow_ext/wrapper.py:40-45, airflow_ext/wrapper.py:118-142
Summary
Scheduler Service Management in airflow-ext provides an automated bridge between Meltano's lifecycle management and Apache Airflow's robust scheduling capabilities. By wrapping the Airflow CLI and providing a dynamic DAG generator, the system allows developers to manage complex data workflows as first-class Meltano objects while benefiting from Airflow's enterprise-grade service architecture.