[scheduler]
min_file_process_interval = 30
parsing_processes = 2
parallelism = 32
max_active_tasks_per_dag = 16
store_serialized_dags = True

[database]
sql_alchemy_pool_size = 20
sql_alchemy_max_overflow = 30
sql_alchemy_pool_recycle = 1800
sql_alchemy_pool_pre_ping = True

[celery]
worker_concurrency = 16
worker_prefetch_multiplier = 1
worker_max_tasks_per_child = 1000
task_acks_late = True

# performance_monitoring.py
from airflow import settings
from airflow.models import DagRun, TaskInstance, DagModel
from sqlalchemy import text, func
from datetime import datetime, timedelta

def get_performance_metrics():
    """Collect comprehensive performance metrics."""
    session = settings.Session()
    
    # Scheduler metrics
    scheduler_metrics = {
        'active_dags': session.query(DagModel).filter(
            DagModel.is_active == True
        ).count(),
        'total_dag_runs': session.query(DagRun).count(),
        'recent_runs_1h': session.query(DagRun).filter(
            DagRun.execution_date >= datetime.now() - timedelta(hours=1)
        ).count(),
    }
    
    # Task metrics
    task_stats = session.query(
        TaskInstance.state,
        func.count(TaskInstance.task_id)
    ).group_by(TaskInstance.state).all()
    
    task_metrics = dict(task_stats)
    
    # Performance metrics
    avg_task_duration = session.query(
        func.avg(TaskInstance.duration)
    ).filter(
        TaskInstance.state == 'success',
        TaskInstance.execution_date >= datetime.now() - timedelta(hours=24)
    ).scalar()
    
    # Error rate
    total_tasks_24h = session.query(TaskInstance).filter(
        TaskInstance.execution_date >= datetime.now() - timedelta(hours=24)
    ).count()
    
    failed_tasks_24h = session.query(TaskInstance).filter(
        TaskInstance.state == 'failed',
        TaskInstance.execution_date >= datetime.now() - timedelta(hours=24)
    ).count()
    
    error_rate = failed_tasks_24h / total_tasks_24h if total_tasks_24h > 0 else 0
    
    return {
        'scheduler': scheduler_metrics,
        'tasks': task_metrics,
        'avg_duration_seconds': avg_task_duration,
        'error_rate_24h': error_rate,
        'timestamp': datetime.now().isoformat(),
    }

def identify_slow_dags(top_n=10):
    """Identify slowest DAGs by average task duration."""
    session = settings.Session()
    
    slow_dags = session.query(
        TaskInstance.dag_id,
        func.avg(TaskInstance.duration).label('avg_duration'),
        func.count(TaskInstance.task_id).label('task_count')
    ).filter(
        TaskInstance.state == 'success',
        TaskInstance.execution_date >= datetime.now() - timedelta(days=7)
    ).group_by(
        TaskInstance.dag_id
    ).order_by(
        func.avg(TaskInstance.duration).desc()
    ).limit(top_n).all()
    
    return [
        {
            'dag_id': row[0],
            'avg_duration': row[1],
            'task_count': row[2],
        }
        for row in slow_dags
    ]

if __name__ == "__main__":
    metrics = get_performance_metrics()
    slow_dags = identify_slow_dags()
    
    print("Performance Metrics:")
    print(f"  Active DAGs: {metrics['scheduler']['active_dags']}")
    print(f"  Avg task duration: {metrics['avg_duration_seconds']:.2f}s")
    print(f"  Error rate (24h): {metrics['error_rate_24h']:.2%}")
    
    print("\nSlowest DAGs:")
    for dag in slow_dags:
        print(f"  {dag['dag_id']}: {dag['avg_duration']:.2f}s avg ({dag['task_count']} tasks)")

from airflow.decorators import task, dag
from datetime import datetime, timedelta
import asyncio

@dag(
    schedule_interval="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['performance', 'optimization'],
)
def optimized_dag():
    
    @task
    def batch_processing():
        """Process data in batches for better performance."""
        import pandas as pd
        
        # Read data in chunks
        chunk_size = 10000
        total_processed = 0
        
        for chunk in pd.read_csv('/data/large_file.csv', chunksize=chunk_size):
            # Process chunk
            processed = chunk.dropna().drop_duplicates()
            total_processed += len(processed)
        
        return {'processed': total_processed}
    
    @task
    def parallel_processing():
        """Process independent tasks in parallel."""
        import concurrent.futures
        
        def process_item(item):
            # Simulate processing
            return item * 2
        
        items = list(range(1000))
        
        # Use thread pool for I/O-bound tasks
        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            results = list(executor.map(process_item, items))
        
        return {'processed': len(results)}
    
    @task
    def cached_computation():
        """Cache expensive computations."""
        from airflow.models import Variable
        import json
        import hashlib
        
        # Check cache
        cache_key = "expensive_computation_result"
        cached = Variable.get(cache_key, default_var=None)
        
        if cached:
            return json.loads(cached)
        
        # Perform expensive computation
        result = sum(i ** 2 for i in range(100000))
        
        # Cache result
        Variable.set(cache_key, json.dumps({'result': result}))
        
        return {'result': result}
    
    batch_processing() >> parallel_processing() >> cached_computation()

optimized_dag()

from airflow.decorators import task, dag
from datetime import datetime
import psutil

@dag(
    schedule_interval="@hourly",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['performance', 'resource-aware'],
)
def resource_aware_dag():
    
    @task
    def check_resources():
        """Check available resources before processing."""
        cpu_percent = psutil.cpu_percent(interval=1)
        memory = psutil.virtual_memory()
        
        return {
            'cpu_available': 100 - cpu_percent,
            'memory_available_percent': 100 - memory.percent,
            'should_process': cpu_percent < 80 and memory.percent < 85,
        }
    
    @task
    def adaptive_processing(resources: dict):
        """Adapt processing based on available resources."""
        if not resources['should_process']:
            return {'status': 'skipped', 'reason': 'insufficient_resources'}
        
        # Adjust batch size based on available memory
        batch_size = int(resources['memory_available_percent'] * 100)
        
        return {
            'status': 'processing',
            'batch_size': batch_size,
        }
    
    resources = check_resources()
    adaptive_processing(resources)

resource_aware_dag()