manutej/luxor-claude-marketplace/apache-airflow-orchestration

Claude Code and agent skill by manutej/luxor-claude-marketplace

Trust tier 3
  • community
Source check pending

Checking that the SKILL.md file is still in the repository.

Complete guide for Apache Airflow orchestration including DAGs, operators, sensors, XComs, task dependencies, dynamic workflows, and production deployment

Install

npx vskill@latest install manutej/luxor-claude-marketplace/apache-airflow-orchestration

SKILL.md

Apache Airflow Orchestration

A comprehensive skill for mastering Apache Airflow workflow orchestration. This skill covers DAG development, operators, sensors, task dependencies, dynamic workflows, XCom communication, scheduling patterns, and production deployment strategies.

When to Use This Skill

Use this skill when:

  • Building and managing complex data pipelines with task dependencies
  • Orchestrating ETL/ELT workflows across multiple systems
  • Scheduling and monitoring batch processing jobs
  • Coordinating multi-step data transformations
  • Managing workflows with conditional execution and branching
  • Implementing event-driven or asset-based workflows
  • Deploying production-grade workflow automation
  • Creating dynamic workflows that generate tasks programmatically
  • Coordinating distributed task execution across clusters
  • Building data engineering platforms with workflow orchestration

Core Concepts

What is Apache Airflow?

Apache Airflow is an open-source platform for programmatically authoring, scheduling, and monitoring workflows. It allows you to define workflows as Directed Acyclic Graphs (DAGs) using Python code, making complex workflow orchestration maintainable and version-controlled.

Key Principles:

  • Dynamic: Workflows are defined in Python, enabling dynamic generation
  • Extensible: Rich ecosystem of operators, sensors, and hooks
  • Scalable: Can scale from single machine to large clusters
  • Observable: Comprehensive UI for monitoring and troubleshooting
DAGs (Directed Acyclic Graphs)

A DAG is a collection of tasks organized to reflect their relationships and dependencies.

DAG Properties:

  • dag_id: Unique identifier for the DAG
  • start_date: When the DAG should start being scheduled
  • schedule: How often to run (cron, timedelta, or asset-based)
  • catchup: Whether to run missed intervals on DAG activation
  • tags: Labels for organization and filtering
  • default_args: Default parameters for all tasks in the DAG

DAG Definition Example:

python
from datetime import datetime
from airflow.sdk import DAG
with DAG(
    dag_id="example_dag",
    start_date=datetime(2022, 1, 1),
    schedule="0 0 * * *",  # Daily at midnight
    catchup=False,
    tags=["example", "tutorial"],
) as dag:
    # Tasks defined here
    pass
Tasks and Operators

Tasks are the basic units of execution in Airflow. Operators are templates for creating tasks.

Common Operator Types:

  1. BashOperator: Execute bash commands
  2. PythonOperator: Execute Python functions
  3. EmailOperator: Send emails
  4. EmptyOperator: Placeholder/dummy tasks
  5. Custom Operators: User-defined operators for specific needs

Operator vs. Task:

  • Operator: Template/class definition
  • Task: Instantiation of an operator with specific parameters
Task Dependencies

Task dependencies define the execution order and workflow structure.

Dependency Operators:

  • >>: Sets downstream dependency (task1 >> task2)
  • <<: Sets upstream dependency (task2 << task1)
  • chain(): Sequential dependencies for multiple tasks
  • cross_downstream(): Many-to-many relationships

Dependency Examples:

python
# Simple linear flow
task1 >> task2 >> task3
# Fan-out pattern
task1 >> [task2, task3, task4]
# Fan-in pattern
[task1, task2, task3] >> task4
# Complex dependencies
first_task >> [second_task, third_task]
third_task << fourth_task
Executors

Executors determine how and where tasks run.

Executor Types:

  • SequentialExecutor: Single-threaded, local (default, not for production)
  • LocalExecutor: Multi-threaded, single machine
  • CeleryExecutor: Distributed execution using Celery
  • KubernetesExecutor: Each task runs in a separate Kubernetes pod
  • DaskExecutor: Distributed execution using Dask
Scheduler

The Airflow scheduler:

  • Monitors all DAGs and their tasks
  • Triggers task instances based on dependencies and schedules
  • Submits tasks to executors for execution
  • Handles retries and task state management

Starting the Scheduler:

bash
airflow scheduler

DAG Development Patterns

Basic DAG Structure

Every DAG follows this structure:

python
from datetime import datetime
from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator
with DAG(
    dag_id="basic_dag",
    start_date=datetime(2022, 1, 1),
    schedule="0 0 * * *",
    catchup=False,
) as dag:
    task1 = BashOperator(
        task_id="task1",
        bash_command="echo 'Task 1 executed'"
    )
    task2 = BashOperator(
        task_id="task2",
        bash_command="echo 'Task 2 executed'"
    )
    task1 >> task2
Task Dependencies and Chains

Linear Chain:

python
from airflow.sdk import chain
# These are equivalent:
task1 >> task2 >> task3 >> task4
chain(task1, task2, task3, task4)

Dynamic Chain:

python
from airflow.sdk import chain
from airflow.operators.empty import EmptyOperator
# Dynamically generate and chain tasks
chain(*[EmptyOperator(task_id=f"task_{i}") for i in range(1, 6)])

Pairwise Chain:

python
from airflow.sdk import chain
# Creates paired dependencies:
# op1 >> op2 >> op4 >> op6
# op1 >> op3 >> op5 >> op6
chain(op1, [op2, op3], [op4, op5], op6)

Cross Downstream:

python
from airflow.sdk import cross_downstream
# Both op1 and op2 feed into both op3 and op4
cross_downstream([op1, op2], [op3, op4])
Branching and Conditional Execution

BranchPythonOperator:

python
from airflow.operators.python import BranchPythonOperator
def choose_branch(**context):
    if context['data_interval_start'].day == 1:
        return 'monthly_task'
    return 'daily_task'
branch = BranchPythonOperator(
    task_id='branch_task',
    python_callable=choose_branch
)
daily_task = BashOperator(task_id='daily_task', bash_command='echo daily')
monthly_task = BashOperator(task_id='monthly_task', bash_command='echo monthly')
branch >> [daily_task, monthly_task]

Custom Branch Operator:

python
from airflow.operators.branch import BaseBranchOperator
class MyBranchOperator(BaseBranchOperator):
    def choose_branch(self, context):
        """
        Run extra branch on first day of month
        """
        if context['data_interval_start'].day == 1:
            return ['daily_task_id', 'monthly_task_id']
        elif context['data_interval_start'].day == 2:
            return 'daily_task_id'
        else:
            return None  # Skip all downstream tasks
TaskGroups for Organization

TaskGroups help organize related tasks hierarchically:

python
from airflow.sdk import task_group
from airflow.operators.empty import EmptyOperator
@task_group()
def data_processing_group():
    extract = EmptyOperator(task_id="extract")
    transform = EmptyOperator(task_id="transform")
    load = EmptyOperator(task_id="load")
    extract >> transform >> load
@task_group()
def validation_group():
    validate_schema = EmptyOperator(task_id="validate_schema")
    validate_data = EmptyOperator(task_id="validate_data")
    validate_schema >> validate_data
start = EmptyOperator(task_id="start")
end = EmptyOperator(task_id="end")
start >> data_processing_group() >> validation_group() >> end
Edge Labeling

Add labels to dependency edges for clarity:

python
from airflow.sdk import Label
# Inline labeling
my_task >> Label("When empty") >> other_task
# Method-based labeling
my_task.set_downstream(other_task, Label("When empty"))
LatestOnlyOperator

Skip tasks if not the latest DAG run:

python
from airflow.operators.latest_only import LatestOnlyOperator
from airflow.operators.empty import EmptyOperator
import pendulum
with DAG(
    dag_id='latest_only_example',
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    catchup=True,
    schedule="@daily",
) as dag:
    latest_only = LatestOnlyOperator(task_id='latest_only')
    task1 = EmptyOperator(task_id='task1')
    task2 = EmptyOperator(task_id='task2')
    task3 = EmptyOperator(task_id='task3')
    task4 = EmptyOperator(task_id='task4', trigger_rule='all_done')
    latest_only >> task1 >> task3
    latest_only >> task4
    task2 >> task3
    task2 >> task4

Operators Deep Dive

BashOperator

Execute bash commands:

python
from airflow.providers.standard.operators.bash import BashOperator
bash_task = BashOperator(
    task_id="bash_example",
    bash_command="echo 'Hello from Bash'; date",
    env={'MY_VAR': 'value'},  # Environment variables
    append_env=True,  # Append to existing env vars
    output_encoding='utf-8'
)

Complex Bash Command:

python
bash_complex = BashOperator(
    task_id="complex_bash",
    bash_command="""
        cd /path/to/dir
        python process_data.py --input {{ ds }} --output {{ tomorrow_ds }}
        if [ $? -eq 0 ]; then
            echo "Success"
        else
            echo "Failed" && exit 1
        fi
    """,
)
PythonOperator

Execute Python functions:

python
from airflow.providers.standard.operators.python import PythonOperator
def my_python_function(name, **context):
    print(f"Hello {name}!")
    print(f"Execution date: {context['ds']}")
    return "Success"
python_task = PythonOperator(
    task_id="python_example",
    python_callable=my_python_function,
    op_kwargs={'name': 'Airflow'},
    provide_context=True
)

Traditional ETL with PythonOperator:

python
import json
import pendulum
from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator
def extract():
    data_string = '{"1001": 301.27, "1002": 433.21, "1003": 502.22}'
    return json.loads(data_string)
def transform(ti):
    # Pull from XCom
    order_data_dict = ti.xcom_pull(task_ids="extract")
    total_order_value = sum(order_data_dict.values())
    return {"total_order_value": total_order_value}
def load(ti):
    # Pull from XCom
    total = ti.xcom_pull(task_ids="transform")["total_order_value"]
    print(f"Total order value is: {total:.2f}")
with DAG(
    dag_id="legacy_etl_pipeline",
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
) as dag:
    extract_task = PythonOperator(task_id="extract", python_callable=extract)
    transform_task = PythonOperator(task_id="transform", python_callable=transform)
    load_task = PythonOperator(task_id="load", python_callable=load)
    extract_task >> transform_task >> load_task
EmailOperator

Send email notifications:

python
from airflow.providers.smtp.operators.smtp import EmailOperator
email_task = EmailOperator(
    task_id='send_email',
    to='recipient@example.com',
    subject='Airflow Notification',
    html_content='<h3>Task completed successfully!</h3>',
    cc=['cc@example.com'],
    bcc=['bcc@example.com']
)
EmptyOperator

Placeholder for workflow structure:

python
from airflow.operators.empty import EmptyOperator
start = EmptyOperator(task_id='start')
end = EmptyOperator(task_id='end')
# Useful for organizing complex DAGs
start >> [task1, task2, task3] >> end
Custom Operators

Create reusable custom operators:

python
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults
class MyCustomOperator(BaseOperator):
    @apply_defaults
    def __init__(self, my_param, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.my_param = my_param
    def execute(self, context):
        self.log.info(f"Executing with param: {self.my_param}")
        # Custom logic here
        return "Result"
# Usage
custom_task = MyCustomOperator(
    task_id="custom",
    my_param="value"
)

Sensors Deep Dive

Sensors are a special type of operator that wait for a certain condition to be met before proceeding.

ExternalTaskSensor

Wait for tasks in other DAGs:

This is the start of the file. Read the full SKILL.md on GitHub.

Security

  • Security scanPassed, 100 of 100Pattern checks on the skill's files.
  • Model reviewPassed

See the full security report

Quality evaluation

  • Task evaluationNot runNo evaluation yet. Run one in Skill Studio.

See evaluation results

Source

Popularity

Repository stars
49
Repository forks
12
7-day trend score
20

Stars and forks are for the whole GitHub repository, not this skill alone. Refreshed 3 Apr 2026, 7 months ago.

Works with

Any agent vskill installs to: 56 targets in all. These 39 have a page on this site.

  • Claude Code
  • Cursor
  • GitHub Copilot
  • Windsurf
  • Codex
  • Gemini CLI
  • Cline
  • Amp
  • Roo Code
  • Goose
  • Continue
  • Kilo Code
  • Junie
  • Trae
  • OpenHands
  • Qwen Code
  • AI Agent
  • Tabnine Chat
  • Sourcegraph Cody
  • Amazon Q
  • Firebase Genkit
  • Sweep
  • Devin
  • Phind
  • Mentat
  • Claude Desktop
  • Aider
  • GPT Pilot
  • Plandex
  • bolt.new
  • v0
  • Replit Agent
  • Anse
  • TypingMind
  • BetterChatGPT
  • LibreChat
  • Warp AI
  • Pieces
  • Fig AI
Embed badge
vskill VERIFIED badge
[![vskill VERIFIED](https://verified-skill.com/api/v1/skills/manutej/luxor-claude-marketplace/apache-airflow-orchestration/badge)](https://verified-skill.com/skills/manutej/luxor-claude-marketplace/apache-airflow-orchestration)

Something wrong with this skill or its page? Tell us.