Skip to main content

Streams

Each file in streams/ defines one stream with three parts.

01requests

Fetch

Pull records from an API, file or database. Sign-in, paging and retries come from the source.

02steps

Shape

Reshape records with DuckDB SQL: filter, join across streams, cast types, compute columns.

03exports

Write

Emit to the output you choose: Singer, files, DuckDB/DuckLake, or a dlt destination.

Requests — fetch​

requests pull raw records. The connector handles sign-in, paging and retries.

requests:
  - name: raw_campaigns
    http:
      path: /v1/accounts/{{ config.account_id }}/campaigns
      method: GET
    # or, for an SDK / database reader:
  - name: raw_orders
    sdk: postgres
    method: query
    arguments:
      query: SELECT id, total::VARCHAR AS total FROM orders WHERE updated_at >= $since
      params: { since: "{{ window.start }}" }

Values are bound parameters ($since) — never pasted into the query text.

Transform — shape​

transform steps reshape records with DuckDB SQL. Mode can run per page (mode: page) or across the whole run (mode: run). Filter, join across streams, cast types, and compute columns.

transform:
  mode: page
  steps:
    - name: cleaned_campaigns
      select: |
        SELECT
          (record->>'id')::BIGINT           AS id,
          record->>'name'                   AS name,
          (record->>'total')::DECIMAL(18,2) AS total,
          record->>'status'                 AS status
        FROM raw_campaigns
        WHERE status = 'ACTIVE'

Steps are plain SQL — no bespoke mapping code. Cross-stream joins let one stream enrich another.

Export — write​

export defines the final dataset contracts and identity keys. Each export becomes a file, table or stream in the output you choose at run time.

export:
  campaigns:
    step: cleaned_campaigns
    primary_key: [id]

Windows and batches​

Incremental streams bind a window so a daily run reads just that day:

requests:
  - name: raw_metrics
    http:
      path: /v1/metrics
      method: GET
      params:
        from: "{{ window.start }}"
        to: "{{ window.end }}"

A batch_size partition binds a list of ids, reading a batch per request.

Continue to outputs.