> ## Documentation Index
> Fetch the complete documentation index at: https://docs.streemlined.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Nodes Reference

> Complete reference for all Streemlined pipeline nodes

This page documents every node available in Streemlined. Nodes are the building blocks of a pipeline and fall into four categories.

| Category       | Nodes                                                                                                                                                                                                                                                            | Purpose                                    |
| -------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------ |
| **Sources**    | [Kafka Consumer](#kafka-consumer), [CSV Source](#csv-source), [JDBC Source](#jdbc-source), [Generic Connect Source](#generic-connect-source)                                                                                                                     | Ingest data into the pipeline              |
| **Processors** | [Transform](#transform), [Branch](#branch), [Merge](#merge), [Explode](#explode), [Lookup](#lookup), [JDBC Request-Reply](#jdbc-request-reply), [Peek](#peek), [Data Masking](#data-masking), [Transform SMT](#transform-smt), [Code Transform](#code-transform) | Transform, route, enrich, and inspect data |
| **Sinks**      | [Kafka Producer](#kafka-producer), [JDBC Sink](#jdbc-sink), [Generic Connect Sink](#generic-connect-sink)                                                                                                                                                        | Output data to external systems            |
| **Utility**    | [Comment](#comment)                                                                                                                                                                                                                                              | Annotate the canvas                        |

***

## Sources

Source nodes ingest data into your pipeline from external systems. Every source has a single output port.

### Kafka Consumer

<Icon icon="kafka" /> Consumes records from one or more partitions of a Kafka topic.

| Property         | Type    | Required    | Description                                                            |
| ---------------- | ------- | ----------- | ---------------------------------------------------------------------- |
| `cluster`        | string  | Yes         | Kafka cluster name (defined in configuration)                          |
| `topic`          | string  | Yes         | Kafka topic to consume from                                            |
| `schemaType`     | enum    | Yes         | `JSON`, `AVRO_SR`, `JSON_SR`, or `PROTO_SR`                            |
| `schemaId`       | number  | Conditional | Schema Registry ID — required for `AVRO_SR`, `JSON_SR`, and `PROTO_SR` |
| `propertiesText` | string  | No          | Extra consumer properties (`key=value`, one per line)                  |
| `stubbed`        | boolean | No          | When `true`, runs as a stub during interactive testing                 |

**Output:** One record per Kafka message, with its schema inferred or fetched from Schema Registry.

<Tip>
  When using Schema Registry, Streemlined automatically fetches and displays the schema in the editor so downstream nodes can offer auto-complete.
</Tip>

```yaml theme={null}
kafka-consumer-1:
  type: kafka-consumer
  cluster: default
  topic: orders
  schemaType: AVRO_SR
  schemaId: 8
```

***

### CSV Source

<Icon icon="file-csv" /> Reads records from a local CSV file — useful for development, testing, and seeding reference data.

| Property     | Type    | Required | Description                                      |
| ------------ | ------- | -------- | ------------------------------------------------ |
| `filePath`   | string  | Yes      | Path to the CSV file                             |
| `separator`  | string  | No       | Field separator (default `,`)                    |
| `skipHeader` | boolean | No       | Treat the first row as a header (default `true`) |

**Output:** One record per CSV row, with field names taken from the header (or positional indices if `skipHeader` is `false`).

```yaml theme={null}
csv-source-1:
  type: csv-source
  filePath: data/customers.csv
  separator: ","
  skipHeader: true
```

<Note>
  CSV Source is primarily intended for local development and testing. For production ingestion, prefer Kafka Consumer or a dedicated connector.
</Note>

***

### JDBC Source

<Icon icon="database" /> Polls rows from a relational database table and ingests them into the pipeline using the JDBC Source Connector.

| Property             | Type   | Required    | Description                                                                                                                     |
| -------------------- | ------ | ----------- | ------------------------------------------------------------------------------------------------------------------------------- |
| `database`           | string | Yes         | Database connection name (defined in configuration)                                                                             |
| `table`              | string | Yes         | Table (or schema-qualified table) to poll                                                                                       |
| `mode`               | enum   | Yes         | `bulk`, `incrementing`, `timestamp`, or `timestamp+incrementing`                                                                |
| `incrementingColumn` | string | Conditional | Numeric monotonic column — required for `incrementing` and `timestamp+incrementing`                                             |
| `timestampColumns`   | array  | Conditional | Timestamp column(s) — required for `timestamp` and `timestamp+incrementing`. When two columns are provided, COALESCE is applied |
| `cluster`            | string | Yes         | Kafka cluster for offset storage                                                                                                |
| `topic`              | string | Yes         | Kafka topic used to store connector offsets                                                                                     |
| `propertiesText`     | string | No          | Extra connector properties (`key=value`, one per line)                                                                          |

| `mode`                   | Behavior                                                                               |
| ------------------------ | -------------------------------------------------------------------------------------- |
| `bulk`                   | Polls the full table on every interval — suitable only for small tables                |
| `incrementing`           | Appends a WHERE clause on an always-growing numeric column to capture new rows         |
| `timestamp`              | Appends a WHERE clause on one or more timestamp columns to capture new or updated rows |
| `timestamp+incrementing` | Combines timestamp and incrementing modes for robust change detection                  |

**Output:** One record per row returned by the connector, with schema fetched from the table or configured in the editor.

```yaml theme={null}
jdbc-source-1:
  type: jdbc-source
  database: default
  table: customers
  mode: incrementing
  incrementingColumn: id
  cluster: default
  topic: streemlined-offsets
  propertiesText: |
    poll.interval.ms=5000
    batch.max.rows=100
```

<Tip>
  Use the **Fetch Schema** button in the editor to auto-populate the output schema from the selected table.
</Tip>

***

### Generic Connect Source

<Icon icon="plug" /> Uses any Kafka Connect source connector to ingest data from external systems.

| Property         | Type    | Required | Description                                                    |
| ---------------- | ------- | -------- | -------------------------------------------------------------- |
| `connectorClass` | string  | Yes      | Fully qualified connector class name                           |
| `propertiesText` | string  | Yes      | Connector configuration (`key=value`, one per line)            |
| `cluster`        | string  | Yes      | Kafka cluster for offset storage                               |
| `topic`          | string  | Yes      | Kafka topic used to store connector offsets                    |
| `stubbed`        | boolean | No       | When `true`, runs as a stub during interactive testing         |
| `errorHandling`  | enum    | No       | `FAIL_PIPELINE` (default), `LOG_SKIP`, or `SEND_TO_ERROR_PORT` |

**Output:** One record per message emitted by the connector, with schema inferred or configured in the editor.

The Generic Connect Source lets you use any Kafka Connect source connector. Place the connector JAR in the `libs/` directory and configure it here.

```yaml theme={null}
generic-source-1:
  type: generic-source
  connectorClass: io.confluent.connect.jdbc.JdbcSourceConnector
  cluster: default
  topic: streemlined-offsets
  propertiesText: |
    connection.url=jdbc:postgresql://localhost:5432/mydb
    connection.user=postgres
    connection.password=secret
    table.whitelist=orders
    mode=incrementing
    incrementing.column.name=id
```

<Tip>
  Use **Fetch Schema** in the editor to contact the connector with the current settings and infer an output schema. See [Connector Plugins](/connector-plugins) for details on installing connectors.
</Tip>

***

## Processors

Processor nodes transform, route, filter, or enrich data as it flows through the pipeline. Each processor has at least one input and one output port.

### Transform

<Icon icon="file-code" /> Transforms records using JSONata expressions.

| Property  | Type   | Required | Description                                    |
| --------- | ------ | -------- | ---------------------------------------------- |
| `mapping` | string | Yes      | JSONata expression defining the transformation |

**Input:** Any record
**Output:** Transformed record based on the JSONata expression

The Transform node is the workhorse of most pipelines. Use it to:

* Map between schemas
* Reshape data structures
* Compute derived fields
* Filter out unwanted fields
* Combine multiple fields

<Tip>
  See [Transformations](/transformations) to learn JSONata syntax and best practices.
</Tip>

```yaml theme={null}
transform-1:
  type: transform
  mapping: |
    {
      "order_id": id,
      "customer_name": customer.name,
      "total": items.price ~> $sum(),
      "processed_at": $now()
    }
```

<Accordion title="JSONata examples">
  **Rename fields:**

  ```jsonata theme={null}
  {
    "userId": user_id,
    "userName": user_name
  }
  ```

  **Flatten nested objects:**

  ```jsonata theme={null}
  {
    "id": order.id,
    "customerEmail": order.customer.email,
    "itemCount": $count(order.items)
  }
  ```

  **Conditional logic:**

  ```jsonata theme={null}
  {
    "tier": totalSpent > 1000 ? "gold" : totalSpent > 500 ? "silver" : "bronze"
  }
  ```

  **Aggregate arrays:**

  ```jsonata theme={null}
  {
    "totalPrice": items.price ~> $sum(),
    "avgPrice": $average(items.price),
    "maxPrice": $max(items.price)
  }
  ```
</Accordion>

***

### Branch

<Icon icon="code-branch" /> Routes records to different outputs based on conditions.

| Property   | Type  | Required | Description                                |
| ---------- | ----- | -------- | ------------------------------------------ |
| `branches` | array | Yes      | List of `{ id, label, condition }` objects |

Each branch contains:

* `id` — Unique identifier
* `label` — Display name
* `condition` — JSONata expression that returns `true` or `false`

**Input:** Any record
**Output:** Multiple outputs — one per branch, plus a `default` output

Records are evaluated against each condition in order. The first matching condition routes the record to that branch's output. Records that match no condition go to `default`.

```yaml theme={null}
branch-1:
  type: branch
  branches:
    - id: high-value
      label: High Value
      condition: total > 1000
    - id: priority
      label: Priority
      condition: customer.tier = "gold"
```

<Warning>
  Conditions are evaluated in order. Place more specific conditions before general ones to avoid unexpected routing.
</Warning>

***

### Merge

<Icon icon="filter" /> Combines multiple parallel input streams into a single output stream (fan-in).

| Property     | Type   | Required | Description                                     |
| ------------ | ------ | -------- | ----------------------------------------------- |
| `inputCount` | number | No       | Number of input ports (default `2`, range 2–10) |

**Inputs:** Multiple inputs — `input-0` through `input-(n-1)`
**Output:** One merged stream on `output`

Input 0 defines the schema for all inputs and the output. Connect the primary stream to input 0; additional inputs must match that schema.

```yaml theme={null}
merge-1:
  type: merge
  inputCount: 3
```

<Note>
  Use Merge after a [Branch](#branch) node to reunite split streams, or to combine records from independent sources that share the same schema.
</Note>

***

### Explode

<Icon icon="layer-group" /> Expands an array field into individual records (a flatMap operation).

| Property         | Type   | Required | Description                        |
| ---------------- | ------ | -------- | ---------------------------------- |
| `arrayToFlatMap` | string | Yes      | Name of the array field to explode |

**Input:** Record containing an array field
**Output:** One record per array element, with the array replaced by its individual items

Use Explode when you need to process array elements individually. For example, if an order contains multiple line items, Explode creates a separate record for each item.

```yaml theme={null}
explode-1:
  type: flatmap
  arrayToFlatMap: items
```

<Accordion title="Before / after example">
  **Before** — one record:

  ```json theme={null}
  {
    "orderId": 123,
    "items": [
      { "sku": "A", "qty": 2 },
      { "sku": "B", "qty": 1 }
    ]
  }
  ```

  **After** — two records:

  ```json theme={null}
  { "orderId": 123, "item": { "sku": "A", "qty": 2 } }
  { "orderId": 123, "item": { "sku": "B", "qty": 1 } }
  ```
</Accordion>

***

### Lookup

<Icon icon="magnifying-glass" /> Enriches records by joining with cached reference data (stream-table join).

| Property                | Type   | Required | Description                                                         |
| ----------------------- | ------ | -------- | ------------------------------------------------------------------- |
| `lookupKey`             | string | Yes      | Expression evaluated on the incoming record to produce the join key |
| `cacheKey`              | string | Yes      | Expression evaluated on the reference data to produce the cache key |
| `fieldName`             | string | Yes      | Name for the enriched field added to the output                     |
| `lookupFailureBehavior` | enum   | No       | `REJECT` (default) or `CONTINUE`                                    |

**Inputs:**

* **input** (left) — Main data stream
* **reference** (top) — Reference data, cached in memory

**Outputs:**

* **output** — Enriched records
* **reject** — Records with no match (only when behavior is `REJECT`)

The reference data is loaded into an in-memory cache keyed by `cacheKey`. For each incoming record, `lookupKey` is evaluated and matched against the cache.

```yaml theme={null}
lookup-1:
  type: lookup
  lookupKey: customer_id
  cacheKey: id
  fieldName: customer_details
  lookupFailureBehavior: REJECT
```

<Note>
  Choose `CONTINUE` if missing reference data is acceptable — the lookup field will be `null`. Choose `REJECT` to route unmatched records to a separate output for error handling or dead-letter queues.
</Note>

***

### JDBC Request-Reply

<Icon icon="database" /> Enriches each record by executing a parameterized SQL query against a relational database.

| Property                | Type    | Required | Description                                                                                   |
| ----------------------- | ------- | -------- | --------------------------------------------------------------------------------------------- |
| `query`                 | string  | Yes      | SQL query with `?` placeholders                                                               |
| `parameters`            | array   | No       | List of `{ expression, reconcileColumn }` objects mapping record fields to query placeholders |
| `fieldName`             | string  | No       | Name for the result field (default `jdbc_result`)                                             |
| `lookupFailureBehavior` | enum    | No       | `REJECT` (default) or `CONTINUE`                                                              |
| `batchSupport`          | boolean | No       | Enable batched query execution for throughput                                                 |

**Input:** Any record
**Outputs:**

* **output** — Record enriched with the query result in `fieldName`
* **reject** — Records where the query returned no rows (only when behavior is `REJECT`)

Unlike [Lookup](#lookup), which joins against a pre-cached dataset, JDBC Request-Reply executes a live query per record (or per batch). This is ideal when reference data is too large to cache or changes frequently.

```yaml theme={null}
jdbc-rr-1:
  type: jdbc-request-reply
  query: "SELECT name, credit_limit FROM customers WHERE id = ?"
  parameters:
    - expression: customer_id
      reconcileColumn: id
  fieldName: customer
  lookupFailureBehavior: CONTINUE
  batchSupport: true
```

<Tip>
  Enable `batchSupport` when enriching high-throughput streams — queries are grouped into batches, significantly reducing round-trips to the database.
</Tip>

***

### Peek

<Icon icon="eye" /> Observes records without modifying them — useful for debugging and monitoring.

| Property   | Type | Required | Description                |
| ---------- | ---- | -------- | -------------------------- |
| `logLevel` | enum | No       | `DEBUG`, `INFO`, or `WARN` |

**Input:** Any record
**Output:** Same record, unmodified

Use Peek to:

* Debug pipeline behavior during development
* Log records at specific points in the pipeline
* Inspect data shapes between processing steps

```yaml theme={null}
peek-1:
  type: peek
  logLevel: INFO
```

<Tip>
  Peek output appears in the console panel in the editor. Use it liberally during development, then reduce log levels or remove Peek nodes before deploying to production.
</Tip>

***

### Data Masking

<Icon icon="shield-halved" /> Masks sensitive fields in each record’s value — redaction, nulling, numeric jitter, or synthetic replacement via [Datafaker](https://www.datafaker.net/) expressions.

| Property       | Type   | Required    | Description                                                                                                                                    |
| -------------- | ------ | ----------- | ---------------------------------------------------------------------------------------------------------------------------------------------- |
| `maskingRules` | array  | Yes         | List of rules. Each rule has `field`, `mask`, and optionally `varianceRange` (for `VARIANCE`) or `fakerExpression` (for `FAKER`)               |
| `schema`       | string | Conditional | JSON string of the **output** Connect/JSON schema — the editor includes this when exporting so the runner can attach the correct `valueSchema` |

**Input:** Any record with a JSON object `value`
**Output:** Same record shape with masked fields applied; if there are no rules, or the value is missing, the record passes through unchanged.

Field paths use dot notation for nested structs. Append `[]` to a segment to apply the rule to **every element** of an array at that path (for example, `items[].email` masks `email` inside each item).

| `mask`     | Behavior                                                                                                                                                     |
| ---------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| `REDACT`   | Replace the value with `***`                                                                                                                                 |
| `NULL`     | Set the field to JSON `null`                                                                                                                                 |
| `VARIANCE` | Add a random delta in `[-varianceRange, +varianceRange]` to numeric fields (integers or floats). If `varianceRange` is omitted or null, it is treated as `0` |
| `FAKER`    | Replace with the result of the Datafaker expression. A blank or missing expression, or a failed evaluation, falls back to `***`                              |

`mask` values are matched case-insensitively at runtime.

```yaml theme={null}
data-masking-1:
  type: data-masking
  maskingRules:
    - field: customer.email
      mask: REDACT
    - field: age
      mask: VARIANCE
      varianceRange: 2
    - field: display_name
      mask: FAKER
      fakerExpression: "#{Name.fullName}"
    - field: order_lines[].internal_id
      mask: NULL
  schema: |
    {"type":"struct","fields":[
      {"field":"customer","type":"struct","fields":[
        {"field":"email","type":"string"}
      ]},
      {"field":"age","type":"int32"},
      {"field":"display_name","type":"string"},
      {"field":"order_lines","type":"array","items":{"type":"struct","fields":[
        {"field":"internal_id","type":"string","optional":true}
      ]}}
    ]}
```

<Note>
  Unknown `mask` values are logged and the field is left unchanged. `VARIANCE` on non-numeric fields is skipped with a warning.
</Note>

<Tip>
  Configure rules in the node editor from the incoming schema. The editor derives the output schema for downstream nodes (for example, `REDACT` / `FAKER` may widen types to string; `NULL` may mark fields optional).
</Tip>

***

### Transform SMT

<Icon icon="wand-magic-sparkles" /> Applies a Kafka Connect Single Message Transform (SMT) to reshape records in flight.

| Property         | Type    | Required | Description                                              |
| ---------------- | ------- | -------- | -------------------------------------------------------- |
| `className`      | string  | Yes      | Fully qualified SMT class name                           |
| `propertiesText` | string  | No       | SMT configuration properties (`key=value`, one per line) |
| `stubbed`        | boolean | No       | When `true`, runs as a stub during interactive testing   |

**Input:** Any record
**Output:** Record transformed by the configured SMT

Use Transform SMT when you need standard Kafka Connect transforms — such as `ExtractField`, `ReplaceField`, or `Cast` — without writing custom code. Configure both input and output schemas in the editor so downstream nodes can validate the result.

```yaml theme={null}
smt-transform-1:
  type: smt-transform
  className: org.apache.kafka.connect.transforms.ExtractField$Value
  propertiesText: |
    field=customer_id
```

<Tip>
  Press Ctrl+Space in the properties editor for autocomplete suggestions based on the selected SMT class.
</Tip>

***

### Code Transform

<Icon icon="code" /> Transforms records using Python or JavaScript code.

| Property   | Type   | Required | Description                         |
| ---------- | ------ | -------- | ----------------------------------- |
| `language` | enum   | Yes      | `python` or `js`                    |
| `code`     | string | Yes      | Transform code evaluated per record |

**Input:** Any record
**Output:** Transformed record based on the code

Each record is passed to your code as `input` (the record value). The last expression or assigned `output` variable becomes the transformed value. Use the built-in code editor to write, test, and debug transforms with sample data.

| Language | Runtime              |
| -------- | -------------------- |
| `python` | GraalPy              |
| `js`     | JavaScript (GraalJS) |

```yaml theme={null}
code-transform-1:
  type: code-transform
  language: python
  code: |
    import datetime
    output = {
      "order_id": input["id"],
      "customer_name": input["customer"]["name"].upper(),
      "processed_at": str(datetime.utcnow())
    }
```

<Tip>
  Open the **Code Editor** from the node properties panel for syntax highlighting, sample input, live execution, and an AI assistant.
</Tip>

<Note>
  Configure input and output schemas in the editor. The output schema defines the shape downstream nodes expect.
</Note>

***

## Sinks

Sink nodes write data from your pipeline to external systems. Every sink has a single input port and no outputs (terminal nodes).

### Kafka Producer

<Icon icon="kafka" /> Produces records to a Kafka topic.

| Property         | Type    | Required    | Description                                                            |
| ---------------- | ------- | ----------- | ---------------------------------------------------------------------- |
| `cluster`        | string  | Yes         | Kafka cluster name (defined in configuration)                          |
| `topic`          | string  | Yes         | Kafka topic to produce to                                              |
| `schemaType`     | enum    | Yes         | `JSON`, `AVRO_SR`, `JSON_SR`, or `PROTO_SR`                            |
| `schemaId`       | number  | Conditional | Schema Registry ID — required for `AVRO_SR`, `JSON_SR`, and `PROTO_SR` |
| `keyExpression`  | string  | No          | JSONata expression for the record key                                  |
| `propertiesText` | string  | No          | Extra producer properties (`key=value`, one per line)                  |
| `stubbed`        | boolean | No          | When `true`, runs as a stub during interactive testing                 |

**Input:** Any record
**Output:** None (terminal node)

```yaml theme={null}
kafka-producer-1:
  type: kafka-producer
  cluster: default
  topic: processed-orders
  schemaType: JSON
  keyExpression: order_id
```

***

### JDBC Sink

<Icon icon="database" /> Writes records to a relational database table.

| Property         | Type    | Required | Description                                             |
| ---------------- | ------- | -------- | ------------------------------------------------------- |
| `table`          | string  | Yes      | Target table name                                       |
| `mode`           | enum    | No       | `INSERT`, `UPSERT`, or `UPDATE`                         |
| `propertiesText` | string  | No       | Extra connection properties (`key=value`, one per line) |
| `stubbed`        | boolean | No       | When `true`, runs as a stub during interactive testing  |

**Input:** Any record (fields must match table columns)
**Output:** None (terminal node)

The JDBC Sink maps record fields to table columns by name. Ensure your upstream transformation produces a schema compatible with the target table.

```yaml theme={null}
jdbc-sink-1:
  type: jdbc-sink
  table: orders
  mode: UPSERT
```

<Warning>
  Column names are case-sensitive. Ensure your field names exactly match your database column names.
</Warning>

***

### Generic Connect Sink

<Icon icon="plug" /> Uses any Kafka Connect sink connector for output.

| Property         | Type    | Required | Description                                            |
| ---------------- | ------- | -------- | ------------------------------------------------------ |
| `connectorClass` | string  | Yes      | Fully qualified connector class name                   |
| `propertiesText` | string  | Yes      | Connector configuration (`key=value`, one per line)    |
| `stubbed`        | boolean | No       | When `true`, runs as a stub during interactive testing |

**Input:** Any record
**Output:** None (terminal node)

The Generic Connect Sink lets you use any Kafka Connect sink connector. Place the connector JAR in the `libs/` directory and configure it here.

```yaml theme={null}
connect-sink-1:
  type: generic-sink
  connectorClass: io.aiven.kafka.connect.http.HttpSinkConnector
  propertiesText: |
    http.url=https://api.example.com/webhook
    http.authorization.type=none
    batching.enabled=true
```

<Tip>
  Check the connector documentation for available configuration options. Streemlined passes configuration directly to the connector. See [Connector Plugins](/connector-plugins) for details on installing connectors.
</Tip>

***

## Utility

Utility nodes help organize and document your pipeline but do not affect data processing.

### Comment

<Icon icon="comment" /> Adds a text annotation to the pipeline canvas.

| Property | Type   | Required | Description                          |
| -------- | ------ | -------- | ------------------------------------ |
| `text`   | string | No       | Comment text displayed on the canvas |
| `width`  | number | No       | Box width in pixels (default `200`)  |
| `height` | number | No       | Box height in pixels (default `100`) |

Comment nodes are visual-only — they have no ports and are ignored at runtime. Use them to:

* Document the purpose of a pipeline section
* Leave notes for teammates
* Mark areas that need future work

<Note>
  Comments are saved as part of the pipeline definition but have no effect on execution.
</Note>

***

## Next Steps

<CardGroup cols={2}>
  <Card title="Transformations" icon="wand-magic-sparkles" href="/transformations">
    Learn JSONata syntax and best practices
  </Card>

  <Card title="Configuration" icon="gear" href="/configuration">
    Configure Kafka clusters, databases, and more
  </Card>

  <Card title="Interactive Testing" icon="flask-vial" href="/interactive-testing">
    Test your pipeline with stubbed sources and sinks
  </Card>

  <Card title="Core Concepts" icon="lightbulb" href="/concepts">
    Understand schemas, ports, and data flow
  </Card>
</CardGroup>