Airflow Celery Worker Error Fix

In this tutorial, you'll learn about Airflow Celery Worker Error Fix. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.

Airflow tasks remain in "queued" state and are never picked up:

Task is in 'queued' state but not executing.

Celery workers are not consuming tasks from the Message Broker. This happens when the broker (Redis/RabbitMQ) is unreachable, workers are not running, the result backend is misconfigured, or the worker cannot deserialize the task.

Step-by-Step Fix

1. Check if Celery workers are running

WRONG — assuming workers are running:

RIGHT — verify:

# List Celery workers
celery -A airflow.executors.celery_executor inspect active

# Check worker processes
ps aux | grep celery

# Check worker logs
tail -f ~/airflow/logs/scheduler/*.log

If no workers are running:

airflow celery worker

2. Verify broker connection

WRONG — broker URL is wrong or unreachable:

[celery]
broker_url = redis://localhost:6379/0
result_backend = db+postgresql://airflow:airflow@localhost:5432/airflow

RIGHT — test the connection:

# Test Redis
redis-cli ping
# Should respond: PONG

# Test RabbitMQ
rabbitmqctl status

# Test Celery connectivity
celery -A airflow.executors.celery_executor inspect ping

If the broker is unreachable:

# Start Redis
redis-server --daemonize yes

# Or fix connection string
export AIRFLOW__CELERY__BROKER_URL=redis://:password@redis-host:6379/0

3. Configure Celery result backend

WRONG — result backend not set or unreachable:

[celery]
result_backend = db+postgresql://airflow@localhost:5432/airflow

RIGHT — use a supported backend:

[celery]
result_backend = db+postgresql://airflow:airflow@localhost:5432/airflow
result_backend_transport_options = {
    "max_retries": 3,
    "interval_start": 0,
    "interval_step": 0.5,
    "interval_max": 3,
}

Initialize the result backend:

celery -A airflow.executors.celery_executor result --backend

4. Match worker concurrency to capacity

WRONG — too many concurrent tasks overwhelm the worker:

[celery]
worker_concurrency = 32  # Too high for a small worker

RIGHT — set based on available resources:

# 2 CPU cores, 4GB RAM
worker_concurrency = 8

# 8 CPU cores, 16GB RAM
worker_concurrency = 16

5. Handle task Serialization issues

WRONG — tasks cannot be serialized for the queue:

def my_func():
    pass

# Serialization error when the function is defined in a module not in PYTHONPATH

RIGHT — ensure tasks are importable:

# Define tasks in files within the dags_folder
# or export PYTHONPATH to include custom modules

# For lambda functions in tasks:
task = PythonOperator(
    task_id="serializable_task",
    python_callable=lambda: print("Hello"),  # Lambdas ARE serializable
)

6. Use Flower for monitoring

# Install Flower
pip install flower

# Start Flower (Celery monitoring UI)
celery -A airflow.executors.celery_executor flower \
    --port=5555 \
    --broker=redis://localhost:6379/0

Open http://localhost:5555 to see:

• Active workers
• Task queues
• Success/failure rates
• Worker resource usage

Expected output: tasks move from "queued" to "running" to "success".

Prevention

• Monitor Celery workers with Flower or Prometheus.
• Set worker_concurrency based on available CPU/memory.
• Use Redis Sentinel or RabbitMQ cluster for broker HA.
• Configure result backend retries for transient failures.
• Restart workers after updating DAG code.

Common Mistakes with celery worker

1. Using foldl instead of foldl' causing stack overflow on large lists
2. Forgetting deriving (Show, Eq) on custom data types needed for debugging
3. Placing the wildcard pattern first in case expressions, making all subsequent patterns unreachable

These mistakes appear frequently in real-world AIRFLOW code. DodaTech's contributors have identified these patterns through analysis of open-source projects and production systems.

Practice Exercise

Write a pure function that safely divides two integers using Maybe, then test it with edge cases like division by zero and negative numbers.

This exercise reinforces the concepts covered in this guide. Try implementing it before checking online solutions.

FAQ