Documentation by ClonePartner

Start typing to search.

Building Pipelines

Reading and writing data

View Markdown

Read from and write to Connectors with call_method, stream pages with backpressure, and batch writes safely.

Reading and writing data

Almost all I/O in a task goes through the call_method step: it calls a method on one of the task's Connectors and feeds the result into the pipeline. This page shows how to read a collection with streaming backpressure, write each record to a destination, and control batching.

Prerequisites

The call_method step

call_method takes three config keys:

Key Required Description
connector Yes Connector slot name from the task's connectors block
method Yes Method name — depends on the connector type
args No Arguments passed to the connector method

See call_method for the full key table and behavior notes.

A source step usually needs no reference to pipeline input, so its config is a static object:

- name: read_users
  type: call_method
  config:
    connector: source
    method: read
    args:
      resource: users

A write step references the current record, so its config is a JSONata string:

- name: upsert_user
  type: call_method
  config: |
    {
      "connector": "destination",
      "method": "upsert",
      "args": {
        "table": "users",
        "data": input,
        "conflict_columns": ["id"]
      }
    }

How collections stream

A step's return value determines what is emitted downstream:

Return value Behavior
undefined Item is dropped — nothing emitted
[] (empty array) Item is dropped — nothing emitted
Array Auto-unwrapped — each element is emitted as a separate item
Anything else Emitted as a single item

Connector methods that return collections (find, read, list, proxy_list) fan out: each record is emitted downstream as its own item, as it arrives. Downstream steps apply backpressure — when a slow write step is at its concurrency limit, upstream reads pause — so the full result set is never held in memory.

Two earlier behaviors are not controlled by the return value: when a step's config evaluates to null/undefined, or when run_if/skip_if skip the step, the original input passes through unchanged.

A complete task reading from one Connector and writing to another:

name: csv-to-postgres
description: Read users from CSV and upsert them into PostgreSQL
output: none
 
connectors:
  source:
    type: csv
  destination:
    type: postgres
 
steps:
  - name: read_users
    type: call_method
    config:
      connector: source
      method: read
      args:
        resource: users
 
  - name: upsert_user
    type: call_method
    concurrency: 5
    config: |
      {
        "connector": "destination",
        "method": "upsert",
        "args": {
          "table": "users",
          "data": input,
          "conflict_columns": ["id"]
        }
      }

read_users emits one item per CSV row. Each row flows to upsert_user independently; concurrency: 5 lets five upserts run in parallel while backpressure keeps the reader in step. output: none suppresses stdout because the destination write is the point of the task.

Note that a call_method step's return value replaces input for the next step — after upsert_user, input is the upsert result, not the original row. If a later step needs a field from the original record, carry it forward in a prior transform and read it via steps.{name}.output.

Pagination

Connectors handle pagination internally. API connectors' collection methods stream page by page and keep fetching until the source is exhausted — for example, the Jira and Freshdesk find methods auto-paginate, and catalog connectors' proxy_list streams all pages by default. Database connectors stream driver cursors the same way (MongoDB find and aggregate iterate the cursor lazily). Your task does not implement paging loops; it consumes one already-streaming source.

Two related task keys exist for cursor persistence across runs: the top-level cursor_store key names a connector slot used to store cursors, and a step may carry a cursor key with a key field. Most read/write tasks do not need either.

For one-task-per-resource design — why a pipeline has exactly one streaming source, and how to combine several pulls in a Job — see the guidance in Task anatomy and authoring workflow.

Choosing a connector method

Each connector type exposes its own method set (find/upsert on SQL databases, read on CSV, list/create on API connectors). The per-connector pages list every method with its args and return type — see the Connectors overview and, for example, PostgreSQL. The validator checks the method name against the connector type's schema and rejects unknown methods at validation time.

Batching writes

Two step keys control how output is grouped before moving downstream:

  • batch collects a step's output items into arrays of N (the final batch may be smaller). Use it when the destination accepts bulk payloads.
  • spool collects all output into a single array, emitted once when the stream ends. Use it only for aggregation — it is unbounded by design.
- name: read_users
  type: call_method
  batch: 100
  config:
    connector: source
    method: read
    args:
      resource: users

Verify it worked

Run the task with --debug and watch each record flow through on stderr:

envoy csv-to-postgres --config connectors.yaml --debug

Then query the destination (in Data Explorer or directly) and confirm the row count matches the source.