Skip to main content

Celery Beat: Scheduling Periodic Tasks

Celery Beat is a task scheduler that periodically enqueues tasks on a schedule. It enables background jobs like sending daily digests, refreshing caches, cleaning up old data, or running reports—all without external cron jobs. Beat runs as a separate process, reads your schedule, and pushes tasks into the Celery queue at the right time.

Understanding Celery Beat​

Celery Beat is a lightweight daemon that maintains an in-memory schedule of tasks. Periodically, it checks if any scheduled tasks are due and enqueues them to the broker, where workers pick them up and execute them. Unlike cron (which requires shell commands and system configuration), Beat is fully integrated with Celery and defined in Python.

Beat requires two components: the Beat service (daemon) and your Celery app with scheduled tasks defined.

Defining Periodic Tasks​

Define schedules in your Celery config using the beat_schedule dictionary:

from celery import Celery
from celery.schedules import crontab

app = Celery('myapp')
app.conf.broker_url = 'redis://localhost:6379/0'

app.conf.beat_schedule = {
    'send-daily-digest': {
        'task': 'tasks.send_daily_digest',
        'schedule': crontab(hour=8, minute=0),  # Every day at 8:00 AM
    },
    'refresh-cache': {
        'task': 'tasks.refresh_cache',
        'schedule': 300.0,  # Every 5 minutes (300 seconds)
    },
    'clean-old-logs': {
        'task': 'tasks.clean_old_logs',
        'schedule': crontab(hour=2, minute=0, day_of_week=0),  # Every Sunday at 2:00 AM
    },
}

@app.task
def send_daily_digest():
    """Send a digest email to all users."""
    users = User.objects.filter(subscribed=True)
    for user in users:
        send_email.delay(user.email, get_digest(user))
    return f'Sent digest to {users.count()} users'

@app.task
def refresh_cache():
    """Refresh expensive cache data."""
    cache.delete('expensive_data')
    return compute_and_cache('expensive_data')

@app.task
def clean_old_logs():
    """Delete logs older than 30 days."""
    cutoff = timezone.now() - timedelta(days=30)
    deleted, _ = LogEntry.objects.filter(created_at__lt=cutoff).delete()
    return f'Deleted {deleted} log entries'

Schedule Types​

Fixed intervals (in seconds):

app.conf.beat_schedule = {
    'every-10-seconds': {
        'task': 'tasks.frequent_task',
        'schedule': 10.0,
    },
}

Crontab (familiar cron syntax):

from celery.schedules import crontab

app.conf.beat_schedule = {
    'every-monday-9am': {
        'task': 'tasks.weekly_task',
        'schedule': crontab(hour=9, minute=0, day_of_week=1),  # Monday at 9:00 AM
    },
    'every-15th-of-month': {
        'task': 'tasks.monthly_billing',
        'schedule': crontab(hour=0, minute=0, day_of_month=15),  # 15th at 00:00
    },
    'every-hour': {
        'task': 'tasks.hourly_sync',
        'schedule': crontab(minute=0),  # Top of every hour
    },
}

Solar events (sunrise, sunset):

from celery.schedules import solar

app.conf.beat_schedule = {
    'at-sunrise': {
        'task': 'tasks.morning_report',
        'schedule': solar('sunrise', lat=40.7128, lon=-74.0060),  # NYC coordinates
    },
}

Passing Arguments and Options​

Pass task arguments and Celery options to scheduled tasks:

app.conf.beat_schedule = {
    'send-digest-with-args': {
        'task': 'tasks.send_digest',
        'schedule': crontab(hour=8, minute=0),
        'args': ('daily',),  # Positional arguments to the task
        'kwargs': {'include_stats': True},  # Keyword arguments
    },
    'process-high-priority': {
        'task': 'tasks.batch_process',
        'schedule': 300.0,
        'options': {
            'queue': 'critical',  # Route to specific queue
            'priority': 9,        # High priority
            'time_limit': 60,     # Max execution time
        },
    },
}

@app.task
def send_digest(frequency, include_stats=False):
    """Send digest email."""
    data = compile_digest(frequency, include_stats)
    return data

@app.task
def batch_process():
    """Batch process with high priority."""
    return process_batch()

Running Celery Beat​

Beat runs as a separate daemon. Start it in a new terminal:

celery -A celery_app beat --loglevel=info

Or combine Beat and worker in a single process (development only):

celery -A celery_app worker --beat --loglevel=info

You should see output like:

[2026-06-02 08:00:00,000] INFO/MainProcess
  Celery beat v5.3.0 is starting.
[beat] Scheduler: Using SchedulingError scheduler
[beat] Scheduler: Initial schedule...
  Executing 'send-daily-digest' (send_daily_digest)
  ...

Timezone Handling​

Beat uses UTC by default. Set timezone to ensure scheduled tasks run at the expected local time:

from celery.schedules import crontab
import pytz

app.conf.timezone = 'America/New_York'

app.conf.beat_schedule = {
    'morning-report': {
        'task': 'tasks.morning_report',
        'schedule': crontab(hour=8, minute=0),  # 8:00 AM EST/EDT
    },
}

Or set globally in your config file (e.g., celeryconfig.py):

import os

CELERY_TIMEZONE = os.getenv('TZ', 'UTC')

Persistent Schedule with Database​

The default scheduler stores the schedule in memory. If Beat crashes, execution history is lost. For production, use a persistent backend:

from django_celery_beat.models import PeriodicTask, IntervalSchedule, CrontabSchedule

# First-time setup: define in code, then create in database
schedule = IntervalSchedule.objects.get_or_create(
    every=5,
    period=IntervalSchedule.MINUTES,
)[0]

PeriodicTask.objects.get_or_create(
    name='refresh-cache',
    task='tasks.refresh_cache',
    interval=schedule,
)

Configure Beat to use Django database:

app.conf.beat_scheduler = 'django_celery_beat.schedulers:DatabaseScheduler'

Now schedules are stored in a database, survive Beat restarts, and can be edited via Django admin without restarting Beat.

Monitoring Scheduled Tasks​

Enable beat_schedule logging in your config:

app.conf.task_track_started = True
app.conf.task_send_sent_signal = True
app.conf.task_acks_late = True

Monitor task execution with Flower:

celery -A celery_app events
# In another terminal:
celery -A celery_app flower
# Visit http://localhost:5555

Key Takeaways​

  • Celery Beat is a lightweight scheduler that enqueues periodic tasks automatically
  • Define schedules in beat_schedule using fixed intervals, crontab, or solar events
  • Run Beat as a separate daemon: celery -A celery_app beat
  • Use timezone-aware scheduling for reliable execution across regions
  • Use a persistent scheduler (django_celery_beat) in production for durability
  • Monitor scheduled task execution with Celery Flower

Frequently Asked Questions​

What's the difference between crontab in Beat and system cron?​

Both use similar syntax, but Beat runs inside your Celery app and directly enqueues Python functions. System cron requires shell scripts and external setup. Beat is simpler, more flexible, and integrates with retries and monitoring.

Can Beat handle task retries automatically?​

No. Beat only enqueues the task. If the task fails, retries are handled by Celery's task logic (the @app.task(max_retries=...) mechanism). Beat doesn't know about failures.

What if Beat and multiple workers are running?​

Beat runs in one place and enqueues tasks globally. All workers listen to the same queue and pick up tasks. If Beat runs in multiple places (not recommended), tasks may be enqueued multiple times—use a persistent scheduler and run Beat on only one server.

Can I dynamically add or remove scheduled tasks?​

Yes, if using a persistent scheduler like django_celery_beat. Update the database, and Beat picks up changes without restarting. With in-memory schedules, restart Beat to apply changes.

How do I test scheduled tasks?​

Call the task function directly in a test: send_daily_digest(). To test scheduling logic, mock the Beat scheduler. Use Celery's current_app.send_task() to verify task enqueuing without running Beat.

Further Reading​