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](/docs/getting-started).

## 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](/docs/sources/kafka) and [Kafka destination](/docs/destinations/kafka) 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**.

| Setting | Value |
| --- | --- |
| id | `tutorial-in` |
| brokers | Your broker address |
| topics | `smeltr-tutorial-in` |

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:

| Target field | Expression |
| --- | --- |
| `message` | `message.toUpperCase()` |
| `event.dataset` | `"smeltr.tutorial"` |

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**.

| Setting | Value |
| --- | --- |
| id | `tutorial-out` |
| brokers | The same broker |
| default topic | `smeltr-tutorial-out` |

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.

Check the result with a script

This Python 3 snippet finds the record from your latest attempt and checks both fields. Run it in the same terminal so it can read `RUN_ID`.

```
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
```

It prints `Verified delivery:` followed by your uppercase message when it finds a match.

| If you don't see the result | What to try |
| --- | --- |
| The file is empty | Check for an error from the producer. Then look at Status → Issues and Logs for connection, permission, or deployment errors. |
| The message isn't uppercase or has the wrong dataset | Check the saved Eval expressions, and make sure you're looking at the right worker group and running commit. |
| You only see older messages | Check for a backlog or worker errors, then read the topic again. If you're using the script, keep the same terminal so `RUN_ID` still identifies the message you sent. |

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.

| Target field | Expression |
| --- | --- |
| `debug` | `null` |
| `_raw` | `null` |

3 records in · 2 records out · debug fields removed

Input 1

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

Kept

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

Keep the order event; remove the debug field.

Input 2

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

Dropped

No record sent to the destination.

Drop the whole record because event\_type is health\_check.

Input 3

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

Kept

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

Keep the error event; remove the debug field.

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:

| Input | What to check |
| --- | --- |
| `"200"` | The result is a number. |
| `"unknown"` | The pipeline handles an invalid value as you intend. |
| A missing field | The pipeline handles an absent value as you intend. |

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](/docs/code-reference) as you build, then follow the [review checklist](/docs/review#check) before deploying.