# Reading and writing data > Source: https://docs.clonepartner.com/building/reading-and-writing-data/ # 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 - A task file and the authoring loop from [Task anatomy and authoring workflow](/building/task-anatomy/) - Connector configurations for your source and destination types ## 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](/reference/steps/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: ```yaml - 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: ```yaml - 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: ```yaml 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](/building/task-anatomy/). ## 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](/connectors/) and, for example, [PostgreSQL](/connectors/postgres/). 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. ```yaml - 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: ```bash envoy csv-to-postgres --config connectors.yaml --debug ``` Then query the destination (in [Data Explorer](/using/data-explorer/) or directly) and confirm the row count matches the source. :::callout{type="warning"} Do not wrap a collection method in the `$callMethod` JSONata function — it buffers the entire result set into memory and defeats streaming. Use a `call_method` step for collections; reserve `$callMethod` for bounded lookups like `find_one` and `count`. See [Working with JSONata](/building/working-with-jsonata/). ::: ## Related - [call_method](/reference/steps/call-method/) — full configuration reference - [Connectors overview](/connectors/) — per-connector method tables - [Transforming and filtering](/building/transforming-and-filtering/) — shape records between read and write - [Idempotent sync and the mapping DB](/building/idempotent-sync/) — make writes safe to re-run