Connect

Run Pipelines on a Schedule

Many pipelines need to run at set times, such as polling a database every 15 minutes or calling an API every weekday morning. The generate input emits a message on a schedule that you set with a duration or a cron expression, and processors such as sql_select or http do the work when each message arrives. Use this cookbook when you want Redpanda Connect to run work on a schedule without a separate scheduler, or when an external scheduler starts a pipeline that runs once.

In this cookbook, you learn to:

  • Schedule a pipeline with a cron expression and a time zone

  • Trigger a database query or an HTTP request from a scheduled message

  • Choose between a pipeline that runs once and one that runs continuously

Prerequisites

Install the Redpanda CLI (rpk) to run Redpanda Connect. See Get Started with Redpanda Connect using rpk for installation instructions.

Some examples also need the following:

  • For the database example, a PostgreSQL database with an orders table that has id, customer, total, and status columns.

  • For the HTTP example, an HTTP endpoint that returns JSON.

To run an example, save it to a file and run rpk connect run <file>.

Schedule messages with the generate input

The interval field of the generate input sets when messages are generated. It accepts either a duration or a cron expression:

Value When messages are generated

A duration, such as 5m

One message when the pipeline starts, then one every five minutes.

A cron descriptor, such as @every 5m, @hourly, or @daily

One message at each scheduled time. The first message waits for the first scheduled time.

A cron expression with five fields, such as */15 * * * *

At each matching minute, at second 0. The fields are minute, hour, day of month, month, and day of week.

A cron expression with six fields, such as */30 * * * * *

At each matching second. The extra first field is seconds.

A cron expression with a TZ=<location name> prefix, such as TZ=Europe/London 0 2 * * *

At each matching time in that time zone. The location name is a name from the IANA Time Zone database.

Cron expressions without a TZ= prefix use UTC, not the time zone of the machine that runs the pipeline.

Run every minute

This pipeline prints a message at the start of every minute:

input:
  generate:
    interval: '* * * * *' # At second 0 of every minute, in UTC
    mapping: |
      root.fired_at = now()

output:
  stdout: {}

Output is similar to the following, one line per minute:

{"fired_at":"2026-10-12T09:34:00.00238+01:00"}
{"fired_at":"2026-10-12T09:35:00.000186+01:00"}

The schedule is in UTC, but the now() function returns the time in the time zone of the machine that runs the pipeline. This output comes from a machine one hour ahead of UTC.

Run at a time in a specific time zone

To schedule work for a local business time, prefix the cron expression with TZ= and a location name. This pipeline runs at 09:00 New York time, Monday to Friday, and follows daylight saving time changes in that time zone:

input:
  generate:
    interval: 'TZ=America/New_York 0 9 * * 1-5' # 09:00 New York time, Monday to Friday
    mapping: |
      root.fired_at = now().ts_format("2006-01-02T15:04:05Z07:00", "America/New_York")

output:
  stdout: {}

Query a database on a schedule

A processor that runs on a scheduled message behaves like a scheduled input. This pipeline reads pending orders from a PostgreSQL table every 15 minutes:

input:
  generate:
    interval: '*/15 * * * *' # Every 15 minutes, in UTC
    mapping: 'root = {}'

pipeline:
  processors:
    - sql_select:
        driver: postgres
        dsn: "${POSTGRES_DSN}"
        table: orders
        columns: [ "id", "customer", "total" ]
        where: status = ?
        args_mapping: 'root = [ "pending" ]'
    # sql_select returns every row in one message as a JSON array.
    # Split the array so each row becomes its own message.
    - unarchive:
        format: json_array

output:
  stdout: {}

Set POSTGRES_DSN to the connection string for your database, for example postgres://<user>:<password>@<host>:5432/<database>?sslmode=disable.

At each scheduled time, the output has one message for each matching row:

{"customer":"acme","id":1,"total":"10.5"}
{"customer":"initech","id":3,"total":"7"}

Each run returns every row that matches the query at that moment. To read only rows that changed since the previous run, filter on a timestamp column in where, or use a change data capture input such as postgres_cdc.

Call an HTTP API on a schedule

To keep the scheduled message and add an API response to it, wrap the http processor in a branch processor. This pipeline calls an API every 10 minutes and stores the response in the status field of the scheduled message:

input:
  generate:
    interval: '0 */10 * * * *' # At second 0 of every tenth minute, in UTC
    mapping: |
      root.fired_at = now()

pipeline:
  processors:
    - branch:
        request_map: 'root = ""'
        processors:
          - http:
              url: "${API_URL}"
              verb: GET
        result_map: 'root.status = this'

output:
  stdout: {}

Set API_URL to the URL of the API. The request_map sends an empty request body, and the result_map copies the response into the message. Output is similar to the following:

{"fired_at":"2026-10-12T09:40:00.000673+01:00","status":{"healthy":true,"queue_depth":3}}

Without the branch processor, the http processor replaces the scheduled message with the response body. For more ways to combine HTTP responses with messages, see Enrichment Workflows.

Run once or continuously

The count field decides whether the pipeline keeps running:

  • With the default count of 0, the generate input never stops, and the pipeline runs until you stop it. Use this when Redpanda Connect runs as a long-lived service and owns the schedule.

  • With count: 1, the generate input sends one message and then shuts down, and the pipeline exits after it processes that message. Use this when another scheduler, such as a Kubernetes CronJob or the system cron daemon, starts the pipeline at the right time.

When an external scheduler starts the pipeline, set interval to an empty string so that the message is generated as soon as the pipeline starts:

input:
  generate:
    count: 1     # Generate one message, then shut down the pipeline
    interval: "" # Generate it as soon as the pipeline starts
    mapping: |
      root.fired_at = now()

output:
  stdout: {}

If you combine count: 1 with a cron expression, the pipeline waits for the next scheduled time, sends one message, and then exits.

If a pipeline that runs continuously restarts, a duration interval sends a message immediately, so the restart adds one run. A cron expression waits for the next scheduled time. The pipeline does not run the schedules that it missed while it was stopped, because the generate input saves no state between runs.

Next steps

Suggested labs

See more
Search all labs