How to Flatten and Reshape Source Streams

TL;DR

Reshape a fetched source stream — explode nested arrays into one row per element, drop duplicate or unwanted rows, reorder, and trim fields — declaratively in config.yaml with stream_transforms, without writing a Python hook.

Prerequisites

Steps

Step 1: Know Where stream_transforms Run

stream_transforms apply to the raw fetched stream, in order, and run before any PRE_TRANSFORM hooks. Hooks and field mappings always see the reshaped data. All transforms are lazy — the stream is only consumed as downstream processing pulls it.

sources:
  users:
    destination: LocalUser
    source_type: custom
    connection: myapp
    stream_transforms:
      - rowmapmany:
          iterable:
            field: row.userGroups
          columns:
            role_id: item.id
            role_name: item.name

Step 2: Explode Nested Arrays with rowmapmany

Use rowmapmany when one source row represents many relationships. The iterable.field expression selects the array; columns maps output columns to Jinja expressions. row refers to the current source record, item to the current array element.

stream_transforms:
  - rowmapmany:
      iterable:
        field: row.resource_access
      columns:
        user_id: row.userId
        role_id: item.id
        role_name: "item.id ~ '-' ~ item.role"

This turns:

{"userId": "u1", "resource_access": [{"id": "R1", "role": "editor"}]}

into:

user_id

role_id

role_name

u1

R1

R1-editor

Expressions are Jinja expressions — no {{ }} braces. Native types are preserved: integers stay integers, lists stay lists, missing values resolve to None. Use ~ to concatenate strings.

Split a delimited string instead of an array with type: split_string:

stream_transforms:
  - rowmapmany:
      iterable:
        field: row.departments
        type: split_string
        delimiter: ","
        strip: true
      columns:
        user_id: row.userId
        department: item

Step 3: Drop Unwanted Rows or Items with skip_if

Add skip_if to drop items (or rows, for rowmap) that match a condition:

stream_transforms:
  - rowmapmany:
      iterable:
        field: row.userGroups
      skip_if: item.id is none
      columns:
        role_id: item.id
        role_name: item.name

A missing item.id becomes None, and None is none is true, so those items are skipped.

Step 4: Keep One Output Row per Input Row with rowmap

Use rowmap (without iterable) for one-to-one reshaping, such as renaming or computing fields:

stream_transforms:
  - rowmap:
      columns:
        unique_id: row.employeeId
        email: row.email
        display_name: "row.firstName ~ ' ' ~ row.lastName"

skip_if on a rowmap drops the whole row.

Step 5: De-duplicate and Reorder

  • distinct removes duplicate rows on the given key. It sorts by the key, so output order is not the input order.

  • sort orders rows by the given fields.

stream_transforms:
  - rowmapmany:
      iterable:
        field: row.userGroups
      columns:
        role_id: item.id
        role_name: item.name
  - distinct: [role_id]
  - sort: [role_name]

Step 6: Keep Only What You Need with select, cut, cutout

  • select keeps only rows where a Jinja expression is truthy: select: row.is_active

  • cut keeps only the listed fields: cut: [role_id, role_name]

  • cutout drops the listed fields: cutout: [raw_groups]

  • head keeps only the first N rows (handy for dry-run checks)

Step 7: Reference for Field Names That Collide with PETL Methods

Rows are passed as PETL Record objects, so row.field works for most field names. If a field name collides with a Record method (for example count, index, or get), use subscript access instead: row['count'].

Example: Keycloak Roles from a User Source

This full example builds one LocalRole row per user group and de-duplicates, entirely in config:

sources:
  users:
    destination: LocalRole
    source_type: custom
    connection: keycloak
    params:
      resource: /users
    stream_transforms:
      - rowmapmany:
          iterable:
            field: row.userGroups
          skip_if: item.id is none
          columns:
            role_id: item.id
            role_name: item.name
            role_owner: row.ownerUserId
      - distinct: [role_id]
      - select: row.role_owner
      - sort: [role_name]

When to Use a Hook Instead

stream_transforms fit any reshape expressible as a fixed pipeline. If the reshape needs conditionals that depend on other sources, deduplication logic that combines fields, or any imperative logic, keep using a PRE_TRANSFORM hook.

Next Steps