Read a CSV file and load each row into RabbitMQ

Turn a structured CSV file into a stream of RabbitMQ messages without writing custom producer code. This blueprint receives a single order record as an input and publishes it to a RabbitMQ (AMQP) exchange using the io.kestra.plugin.amqp.Publish task. Paired with a fan-out parent flow, it lets you ingest an entire dataset row by row into a message queue, solving the common problem of bridging batch files (exports, reports, order dumps) into event-driven, queue-based consumers.

How it works

1. The flow declares a single order input of type STRING that carries one record (the URI of a one-row ION file produced by the parent flow).
2. The publish_to_rabbitmq task of type io.kestra.plugin.amqp.Publish connects to the broker via the host AMQP URI and targets the exchange named test-queue.
3. The from property reads the record with {{ read(inputs.order) }} and publishes its content as a message to the exchange.

To load a full file, run a parent flow that downloads the CSV, converts it, and fans out one execution per row:

id: read_orders
namespace: company.team

tasks:
  - id: csv
    type: io.kestra.plugin.fs.http.Download
    uri: "https://huggingface.co/datasets/kestra/datasets/raw/main/csv/orders.csv"

  - id: csv_to_ion
    type: io.kestra.plugin.serdes.csv.CsvToIon
    from: "{{ outputs.csv.uri }}"

  - id: each
    type: io.kestra.plugin.core.flows.ForEachItem
    items: "{{ outputs.csv_to_ion.uri }}"
    batch:
      rows: 1
    namespace: company.team
    flowId: produce_to_rabbitmq
    wait: true
    transmitFailed: true
    inputs:
      order: "{{ taskrun.items }}"

What you get

• A reusable producer that publishes any single record to a RabbitMQ exchange.
• Row-level ingestion from a CSV file into a message queue.
• A clean separation between reading data (parent flow) and producing messages (this flow).
• Per-row execution visibility so you can see exactly which records succeeded or failed.

Who it's for

• Data engineers bridging batch exports into event-driven pipelines.
• Backend developers seeding RabbitMQ queues from structured files.
• Platform teams standardizing how files become queue messages.

Why orchestrate this with Kestra

RabbitMQ moves messages but does not schedule, source, or transform the data that feeds it. Kestra fills that gap. You can trigger ingestion on a schedule or on an event, retry failed publishes automatically, and fan out one execution per row with ForEachItem so every record gets its own traceable run with full lineage. The entire pipeline is declarative YAML, version controlled and reviewable, instead of a bespoke producer service you have to deploy and babysit.

Prerequisites

• A running RabbitMQ broker reachable at the AMQP host URI.
• An exchange (here test-queue) the producer can publish to.

Secrets

This blueprint uses inline demo credentials (guest/guest) in the host URI and references no Kestra secrets. For production, move the broker URL and credentials into a secret, for example {{ secret('RABBITMQ_URL') }}.

Quick start

1. Start RabbitMQ locally with Docker:

docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:latest

2. Open the management UI at http://localhost:15672/ and log in with guest/guest.
3. Add this flow and the read_orders parent flow to your namespace.
4. Execute read_orders. It fans out one produce_to_rabbitmq execution per CSV row, each publishing a message to the exchange.

How to extend

• Swap the exchange and add a routing key to target specific queues.
• Point the parent flow at your own CSV or another serdes source (JSON, Avro, Parquet).
• Increase batch.rows to publish multiple records per execution.
• Add retries and an errors branch to alert on failed publishes.
• Replace inline credentials with a secret and parameterize the broker host per environment.

Links

• Kestra AMQP plugin