Transforming and filtering

Between reading and writing, most tasks reshape records and drop the ones they do not need. Three steps cover this: transform evaluates a JSONata expression per item, filter keeps or drops items on a boolean predicate, and map_fields applies a declarative field mapping built for migrations.

Prerequisites

• A task with a streaming source — see Reading and writing data
• JSONata basics from Working with JSONata

transform: reshape each item

The config of a transform step must be a JSONata expression string. It is evaluated per item with the input, arguments, steps, and variables context, and the result becomes the item emitted downstream.

name: extract-contacts
description: Read contacts from CSV and reshape them
connectors:
  source:
    type: csv
steps:
  - name: read_contacts
    type: call_method
    config:
      connector: source
      method: read
      args:
        resource: contacts
 
  - name: extract
    type: transform
    config: |
      {
        "id": input.id,
        "full_name": input.first_name & " " & input.last_name,
        "email": input.email
      }

Result semantics:

• An array result is auto-unwrapped — each element is emitted separately.
• An [] result drops the item.
• A null/undefined result passes the original input through unchanged — it does not drop the item. A pattern like condition ? input : undefined therefore keeps non-matching items. Use a filter step to remove items.

See transform for the full reference.

filter: keep or drop items

The config of a filter step is a boolean JSONata expression string, evaluated per item:

• truthy — the item passes through unchanged.
• falsy (including false, null, 0, "", or a missing field) — the item is dropped; nothing is emitted downstream.

- name: nonzero
  type: filter
  config: "input.balance > 0"

This is the first-class way to remove items from a stream. Prefer it over a transform that returns undefined (which passes through, per the footgun above) and over run_if/skip_if (which skip a step's processing but still pass the item to the next step). See filter.

map_fields: declarative mapping for migrations

map_fields builds a new record from a source record using a mapping definition — one entry per source field. It is designed for migrations where a source schema must be translated into a destination schema, and where the mapping itself is data you want to review, diff, or load from a file.

Config keys:

One of mapping_file or mapping is required. Each mapping entry is either a plain destination field name (simple copy) or an object with field (destination name), and optional transform (JSONata producing the value), skip (JSONata — skip this field when true), path (OData navigation binding), type (number, date, or string coercion), and format (date format used with type: date). A mapping file can also carry default_attributes (values merged into every output record when not already set) and skip_record (a JSONata preprocess/postprocess pair that drops whole records). See map_fields for details.

Prefer map_fields over transform when the work is field-by-field translation: many fields with per-field transforms and skip conditions stay readable as a mapping table, and a mapping_file lets you change the mapping without touching the task. Prefer transform when the output is computed from the record as a whole — restructuring, merging step outputs, or building nested payloads.

Worked example: chaining the three steps

name: stage-active-contacts
description: Read contacts, keep active ones, map them to the destination schema
output: none
 
connectors:
  source:
    type: csv
  staging:
    type: sqlite
 
steps:
  - name: read_contacts
    type: call_method
    config:
      connector: source
      method: read
      args:
        resource: contacts
 
  - name: active_only
    type: filter
    config: 'input.status = "active"'
 
  - name: normalize
    type: transform
    config: |
      {
        "id": input.id,
        "full_name": input.first_name & " " & input.last_name,
        "email": input.email,
        "company": input.company
      }
 
  - name: map_contact
    type: map_fields
    config:
      mapping:
        full_name: FullName
        email: EMailAddress1
        company:
          field: CompanyName
          transform: "$trim(record.company)"
          skip: "record.company = null"
 
  - name: write_contact
    type: call_method
    config: |
      {
        "connector": "staging",
        "method": "upsert",
        "args": {
          "table": "contacts",
          "data": input,
          "conflict_columns": ["FullName"]
        }
      }

The item shape at each stage:

1. After read_contacts: one raw CSV row per item, for example { "id": "42", "first_name": "Ada", "last_name": "Byron", "email": "[email protected]", "company": " Analytical ", "status": "active" }.
2. After active_only: the same shape — rows with any other status are gone from the stream.
3. After normalize: { "id": "42", "full_name": "Ada Byron", "email": "[email protected]", "company": " Analytical " }.
4. After map_contact: { "FullName": "Ada Byron", "EMailAddress1": "[email protected]", "CompanyName": "Analytical" } — destination field names, trimmed company, and the field skipped entirely when company is null.

Note that inside map_fields expressions the source record is referenced as record (the mapping context), not input.

Choosing where the work happens

• Filter at the source when the connector can. A find with a filter arg or a list method with query params moves less data than pulling everything and filtering in the pipeline. Use a filter step when the source method cannot express the condition, or when the predicate depends on earlier step output.
• transform for computed shapes. String concatenation, conditionals, restructuring, and payload building belong in a transform.
• map_fields for schema translation. When the essence is "source field X becomes destination field Y, with these per-field rules", a mapping stays maintainable where an equivalent transform grows unreadable — and mapping_file keeps large mappings out of the task YAML.
• run_if/skip_if for conditional execution, never for dropping. They skip one step's processing; the item continues downstream unchanged.

Verify it worked

Run with --debug and check the emitted item after each step against the shapes above:

envoy stage-active-contacts --config connectors.yaml --debug

The count of items after the filter step should match the number of active records in the source.

Related

• transform — full step reference
• filter — full step reference
• map_fields — mapping patterns, lookups, and mapping files
• Working with JSONata — the expression language behind all three steps