# Check for import errors
airflow dags list-import-errors

# Validate DAG file
python dags/my_dag.py

# Check scheduler logs
docker logs airflow-scheduler-1 | grep -i error

# Check worker status
airflow celery status

# Verify executor configuration
airflow config get-value core executor

# Check for resource constraints
kubectl top pods -n airflow

# Use external storage for large data
def store_large_result(**context):
    # Store in S3 instead of XCom
    s3_hook.load_string(large_data, key='results/data.json', bucket='my-bucket')
    return 's3://my-bucket/results/data.json'  # Return reference only

# Tune scheduler settings
AIRFLOW__SCHEDULER__PARSING_PROCESSES: 4
AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL: 30
AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL: 60

# Add detailed logging
import logging
logger = logging.getLogger(__name__)

def my_task(**context):
    logger.info(f"Starting task with context: {context}")
    # ... task logic
    logger.debug(f"Intermediate result: {result}")

# Test specific task
airflow tasks test my_dag my_task 2026-01-15

# Clear task state for re-run
airflow tasks clear my_dag -t my_task -s 2026-01-15 -e 2026-01-15

# Trigger DAG run
airflow dags trigger my_dag --conf '{"key": "value"}'