Tutorial

Your first pipeline

Read JSON records from one Kafka topic, uppercase each message, add a field, and send them to another topic. You'll build it all from one route and review the change before it reaches a worker.

You'll need a running Smeltr installation, administrator access, and an online worker in a group you can use for testing. If you're still setting up, start with the setup guide.

1. Before you start

We'll use two Kafka topics: one for incoming records and one for the result. Keeping them separate prevents the output from feeding back into the pipeline.

In a terminal with Kafka's command-line tools available, replace localhost:9092 with your test broker's address and create the topics:

export BROKERS='localhost:9092'
kafka-topics.sh --bootstrap-server "$BROKERS" --create \
  --topic smeltr-tutorial-in --partitions 1 --replication-factor 1
kafka-topics.sh --bootstrap-server "$BROKERS" --create \
  --topic smeltr-tutorial-out --partitions 1 --replication-factor 1

These commands assume a single test broker with a plaintext listener. For an authenticated broker, see the Kafka source and Kafka destination guides. If the topics already exist, make sure they're yours to use for this exercise.

Send three records to the input topic, so there is something to capture:

printf '%s\n' \
  '{"service":"checkout","message":"payment declined"}' \
  '{"service":"search","message":"query timed out"}' \
  '{"service":"checkout","message":"order placed"}' |
kafka-console-producer.sh --bootstrap-server "$BROKERS" --topic smeltr-tutorial-in

2. Start a route

A route takes records from a source, runs them through a pipeline, and delivers them to destinations. You'll make all three from the route itself.

In Smeltr, select your test worker group and open Build. It opens on Routes. Choose + Add route. The new route shows three pickers: from source, pipeline and destination.

3. Add the source

In from source, choose + New source…. Choose Pull, all workers, then Kafka. Fill in the settings below and choose Add.

Use broker addresses the worker can reach, including the addresses Kafka advertises to clients. The worker needs permission to read the input topic and write to the output topic.

Smeltr saves the source at once. The first save asks for a branch, because main is only changed by deploying. Call it tutorial-first-pipeline and choose Edit on this branch. Everything you save from here on goes on that branch.

4. Make the pipeline from a capture

In pipeline, choose + New pipeline…. The pipeline tutorial-in-parse opens and asks for a capture of your source. A capture reads records through a worker without deploying anything.

Open Advanced, set Start at to earliest, and choose Start. When the three records appear, choose Stop capture. Enter tutorial-message under Save as and choose Save sample.

Smeltr reads the sample and proposes the steps. It reports that 3 of 3 records read as JSON, and adds a Parser step that declares service and message. The parsed records show beside the steps.

5. Shape the records

After the Parser step, choose + step, then Eval. An Eval step sets fields to the values of expressions. Fill in the first row, then choose + assignment for the second:

The first row uppercases the parsed message. The second labels it as part of this tutorial. A dot in the field name puts dataset inside an event object. The output updates as you type. Its first record should have these fields, along with any it kept from the input:

{
  "service": "checkout",
  "message": "PAYMENT DECLINED",
  "event": {
    "dataset": "smeltr.tutorial"
  }
}

Choose Save. Smeltr saves the pipeline and takes you back to the route, with tutorial-in-parse picked.

6. Add the destination

In destination, choose + New destination…, then Kafka. Fill in the settings below and choose Add.

The route now reads tutorial-in → tutorial-in-parse → tutorial-out. Choose Save to save the route.

7. Check where records go

Open the Route a sample tab. Select tutorial-message and choose Run. Smeltr runs the sample through your routes. Nothing is delivered.

All three records should reach tutorial-out, with none unrouted or dropped. Choose the tutorial-out count to see the documents as they would be sent.

8. Review and deploy

Nothing you've saved runs until it's deployed. In the branch bar, choose Open for review. Smeltr fills in the title from what changed. Give it a description and choose Open for review.

On the review page, read through Saves and make sure it holds only your tutorial changes.

If someone has changed main while you were working, choose Bring current and check the saves again. Then choose Approve & deploy and confirm. Smeltr opens Status, where you can watch your workers pick up the change. If a worker reports a deployment failure, resolve it before sending a record.

9. Send a real record

Back in the same terminal, send a fresh message to the input topic. The timestamp in it will help you tell it apart from earlier attempts.

export RUN_ID="smeltr-tutorial-$(date +%s)"
printf '{"service":"checkout","message":"%s"}\n' "$RUN_ID" | kafka-console-producer.sh \
  --bootstrap-server "$BROKERS" --topic smeltr-tutorial-in

Read the output topic into a file, then open it. The command stops after 30 seconds without another record.

kafka-console-consumer.sh --bootstrap-server "$BROKERS" \
  --topic smeltr-tutorial-out --from-beginning \
  --timeout-ms 30000 > delivered.jsonl
cat delivered.jsonl

Look for the message you just sent, now in uppercase, with event.dataset set to smeltr.tutorial. The three records you sent first are there too: the source starts at the earliest record. Open Map to see your route with records flowing through it.

python3 - <<'PY'
import json, os
expected = os.environ["RUN_ID"].upper()
with open("delivered.jsonl") as records:
    found = any(
        row.get("message") == expected
        and row.get("event", {}).get("dataset") == "smeltr.tutorial"
        for line in records if line.strip()
        for row in [json.loads(line)]
    )
assert found, f"Expected record not delivered: {expected}"
print(f"Verified delivery: {expected}")
PY

When you're finished, remove the tutorial route, pipeline, sample, source and destination in a new reviewed change. Once the workers are running that change, you can remove the two Kafka topics too.

Try filtering rows and fields

Use a separate pipeline to try this example. Keep order and error events, drop routine health checks, and remove the debug field. Copy each input below into its own saved sample.

1. Add a JSON Parser step with _raw as its input. Declare event_type, message, and debug as str fields.
2. Add a Drop step with condition event_type == "health_check". A true condition removes the whole record.
3. Add an Eval step with the assignments below. Setting these fields to null leaves them out of the delivered JSON. _raw is the original input text.

{
  "event_type": "order",
  "message": "Order placed",
  "debug": "trace-101"
}

{
  "event_type": "order",
  "message": "Order placed"
}

{
  "event_type": "health_check",
  "message": "Service is healthy",
  "debug": "trace-102"
}

{
  "event_type": "error",
  "message": "Payment timed out",
  "debug": "trace-103"
}

{
  "event_type": "error",
  "message": "Payment timed out"
}

Run each sample and check the result shown above.

Try your own data

Pick a few records from your source and remove sensitive values before saving them as samples. Work on a branch and check which routes use the pipeline; an edit to shared logic can affect other worker groups. Add one step at a time and inspect the fields and types in the preview.

Include records that are awkward to handle. If your next pipeline converts a status code to a number, try these cases:

Test filtering in both directions too. For a Drop step, save one record the step should drop and another that should pass through. When removing fields, check that unwanted columns are absent and the fields you need are still present.

The preview checks your transformation; send a real record to check destination access too.

Use the expression reference as you build, then follow the review checklist before deploying.