ETL Pipelines

When a pipeline is the right tool#

AgentSwarms has three ways to reshape data, and they are deliberately not the same feature. Data preparation transforms tables the platform can already query, in place. The Semantic Layer defines governed metrics over warehouse tables. An ETL pipeline moves data between systems: it exists for the work the other two cannot express — calling an external API, converting raw bucket files, reading one database and writing another, or any transform that wants a real programming language.

Building one#

The visual canvas#

A pipeline is a graph: source nodes on the left, transform nodes in the middle, target nodes on the right, connected by dragging between handles. Multiple sources can feed a join or a union; one output can branch to several targets. Click a node to configure it in the side panel; the Code toggle shows the exact Python the graph compiles to — what you see is literally what a run executes. A graph that isn't valid yet (a join with one input, an unconnected node) still saves; the canvas shows what to fix, and the run button refuses until it compiles.

Sources

A Data Catalog asset (any table, view, file or dataset the catalog crawled), object storage files (CSV, TSV, JSON, JSONL, Parquet, Excel — the datasets the crawl found, or a glob like raw/orders/*.csv), a database or warehouse table or SQL query, an HTTP API returning JSON, a platform dataset (uploads, prep outputs, connector-synced tables), a Lakehouse table, a Kafka, Kinesis or Pub/Sub stream, or custom Python.

Transforms

Filter, select, rename, derive, join (inner/left/right/outer), union, aggregate (group by with sum/mean/min/max/count/…), sort, deduplicate, fill or drop nulls, limit, quality gate (validate columns; fail, warn or drop on violation), SQL over the incoming frame, or custom Python.

Targets

Object storage (Parquet, CSV or JSONL under <bucket>/<dataset>/<table>/) or a database/warehouse table — each with replace, append or merge (upsert on primary keys). SQL-family databases load over SQLAlchemy; Snowflake, BigQuery and Databricks load through their native bulk paths using the connection you already added. Bucket targets can also write Delta Lake or Iceberg tables instead of plain files, and a Lakehouse target writes ACID, snapshotted tables into the built-in warehouse.

Nothing is typed that the platform already knows. A connection, then its schema, then a table; a bucket, then a folder, then a dataset; a lakehouse schema, then a table; a catalog source, then its schema or folder, then an asset; secrets by name; AWS regions. Every one is a picker, each level narrowing the next, and a target that may create something offers New … before asking for a name. Each pick reports its columns, so the incremental cursor, the merge keys and the transforms downstream are picked too, before any preview has run. What stays a field is what only you know: a URL, a topic, an expression, a query.

Code#

A full-height Python editor. The contract is small: define entrypoint(inputs=None), return a JSON-able metrics dict (at minimum {'rows_loaded': n}), read credentials from environment variables, and list the packages you import under Requirements — they are installed in the sandbox before the run. Ejecting a canvas pipeline hands you its compiled Python as a starting point (one-way: edited code cannot be turned back into a graph).

AI generate and refine#

The AI assist panel drafts a pipeline from a brief, or refines the current code from an instruction (“add retry logic to the fetch”, “partition by month”). It uses your connected provider and the shared model picker, so IAM model rules apply here exactly as in BI, and drafts follow the runtime contract — credentials from the environment, never literals. A draft is only ever text in the editor until you review, save and run it.

Sample pipelines#

The New pipeline dialog offers six worked scenarios, each solving a classically hard ETL problem against messy demo datasets bundled with the app (under /etl-samples/ on your own deployment — the runs fetch them from your instance's origin, so they work offline). Pick your destination bucket under Settings and run.

Connections and supported systems#

Pipelines reuse the connections the rest of the product already governs — object-storage sources from the Data Catalog (AWS S3, MinIO, R2, Spaces, B2, GCS in S3-compat mode) and database connections from Data Sources, including ones shared with you through IAM. Credentials are resolved server-side at run start and delivered into the sandbox process only.

How a run executes#

Every run is a batch kernel on the sandboxed runtime — the same hardened containers, egress allow-list and reaper as the Developer workspace. Nothing runs in the app process. The run installs the pipeline's requirements, fetches its resolved credentials over HTTPS into process memory (never container environment variables, never the code text), executes entrypoint(inputs), and reports metrics, logs and status back to the Runs tab. Secret values are scrubbed from captured logs before they are stored.

Engines: the sandbox, or a Spark cluster#

Every pipeline runs on the pandas engine unless it says otherwise: one sandbox, an in-memory pandas program, the sizing table below. That is right for most pipelines, and nothing about it changed. A pipeline whose data does not fit one box picks the Spark engine in Settings → Engine. The graph, the canvas, the run log and the metrics are the same; what changes is where the program executes: the compiler emits PySpark instead of pandas, and the sandbox drives a cluster over Spark Connect — it holds only the pure-Python client, no JVM, so every hardening decision made for it stands. Previews always sample in the sandbox.

Where pandas differs from SQL the pandas behaviour is reproduced on purpose — a null group key forms a group, nulls sort last either way, a join suffixes shared columns _x/_y, a null fails a range or regex check — so a pipeline gives the same answer on either engine. Filter and derive expressions keep their pandas spelling and are translated to Spark SQL at save time; a construct Spark cannot express is refused at save, naming it and the fix. The engine also refuses, at save: Iceberg targets (write Delta), merge into plain files (merge needs a Delta table), merge into a database, and a SQL step calling regexp_extract without its third argument — DuckDB returns the whole match there while Spark reads the missing argument as capture group 1, so the step is asked which it means rather than left to fail minutes into a cluster run.

Where the cluster comes from is one choice for the deployment, in Admin → Developer runtime → Spark engine. An endpoint you run is the default: one Spark Connect endpoint shared by every run, set there or as SPARK_CONNECT_URL — and until it is set the engine picker says so. Locally, docker compose up -d gives you one at sc://spark-connect:15002 with the S3A, Delta and JDBC connectors already on it.

One cluster per run, on Kubernetes is the other, available when the app runs in a cluster. Each run gets its own driver — which is also its Spark Connect endpoint — and the executors it asks for, sized in the admin form; both are deleted when the run ends, so a run's size is the node pool rather than one box and nothing is paid for between runs. Apply deploy/k8s/spark/spark-runtime.yaml first for the namespace, the service account, the quota and the network policy. Because provisioning takes minutes on a stock image, a run stays queued while its cluster comes up; bake the connector jars into your own image and set SPARK_PACKAGES= to make that fast.

Credentials and secrets#

Storage nodes resolve their catalog source's credentials; database nodes resolve their connection's. Each node's variables are namespaced by its id (visible in the generated code), and storage access stays scoped to the source's configured bucket prefix. Code-mode pipelines get the pipeline-level destination as:

Anything else the code needs — an API token, a password for a system without a stored connection — is bound under Settings as KEY={{secret:NAME}} lines referencing Secrets, and arrives the same way. A binding that fails to resolve is dropped rather than failing the run, so the code that needed it reports a missing variable — far easier to diagnose than a run that never starts.

Schedules, triggers and chaining#

• Manual — the Run now button, or Run with parameters beside it (see below).
• Hourly / daily / weekly — simple presets, swept by the same scheduler that drives BI refreshes and catalog crawls. A pipeline that overruns its interval skips a beat rather than queueing a backlog behind itself.
• Cron — a five-field expression evaluated in an IANA timezone: 0 6 * * 1-5 with Europe/Berlin runs weekdays at 06:00 Berlin time, DST handled. Supported syntax: *, numbers, ranges, lists and steps (*/15); typos are refused at save with the reason.
• Continuous — one long-running run drains a stream source (Kafka, Kinesis, Pub/Sub), webhook ingest, CDC or an incremental cursor every few seconds, persisting its position after every committed load, and the sweep restarts it whenever none is live. Stop it from the card. When every target is a lakehouse table the card says exactly-once: a tick's loads and its positions commit in one lakehouse transaction and the next run resumes from what committed with the rows, so a crash cannot replay a batch. Rollover and restart backoff: ETL_CONTINUOUS_ROLLOVER_MINUTES, ETL_CONTINUOUS_RESTART_BACKOFF_SECONDS.
• External trigger — mint a token under Settings and POST /api/etl/run with it as a bearer plus {"pipeline_id": "…"} (optionally "params"). This is how a swarm's http node, an n8n workflow or CI starts a load. Tokens are shown once and stored only as a hash; triggering is rate-limited (ETL_TRIGGER_PER_MIN, default 6/min per pipeline).
• Chaining — set Run after in Settings and this pipeline starts when the selected one succeeds (medallion layers, staging → merge). Cycles are refused at save time.

Beyond pipelines: SQL models and ML schedules#

Run after chains a pipeline to another pipeline. A pipeline can also say what to start when a run succeeds, in Settings → “After it succeeds, also…”: build SQL models — every active model you own in dependency order, or the ones you pick with everything they depend on built first — and run ML schedules, any retrain or batch-predict schedule you own. Both run as the pipeline's owner, and both are their own runs on their own pages (the build carries the trigger chain), so a failure there never changes the pipeline's outcome. The save path refuses a model or schedule that is not yours, by name.

Retries and overlap#

A pipeline can retry up to five times with exponential backoff (1, 2, 4, 8, 16 minutes). Retries reuse the same run, so the Runs tab shows one logical run whose logs carry every attempt's story; a runtime that was briefly unavailable counts as a retryable failure, and the failure notification fires only when the ladder is exhausted. Overlap is refused by default — a second start (manual, webhook, chain or schedule) is rejected while a run is queued, running or waiting out a backoff, because append-mode targets double-load under overlap. Flip Allow concurrent runs when a pipeline is genuinely idempotent.

Parameters and backfills#

Every run delivers a JSON object to entrypoint(inputs): the pipeline's default parameters merged under any per-run values from the Run with parameters dialog or the trigger body. The object is pinned on the run row, so “re-run July 3–9” is one dialog away and forever attributable:

def entrypoint(inputs=None):
    inputs = inputs or {}
    start = inputs.get('start_date', '2026-08-01')
    end = inputs.get('end_date', '2026-08-07')
    # query/filter the window, load as usual

Engine-managed incremental loads#

Database and object-storage source nodes take an optional incremental cursor column. The engine stores the highest value each successful run loaded and hands it back to the next run; database sources push the filter down as SQL, storage sources filter rows after reading. The watermark advances only after a durable load — a crash between load and bookkeeping re-reads rows rather than skipping them — and an empty read keeps the previous cursor. State is server-held per node and never client-writable.

An object-storage source can also be a landing zone: turn on Only files not loaded before and every run lists the prefix and reads only the files the engine has not loaded, keeping a small ledger on the cursor (the newest modification time loaded and the keys at that time). Files per run bounds a backlog; a re-uploaded file loads again as a new version, so pair it with a merge target when that matters. The source becomes drainable, which is what lets the pipeline run continuous — exactly-once into the lakehouse. Sandbox engine only; the Spark engine reads a prefix whole and refuses the setting at save.

What the cluster writes, and what it does not#

On the Spark engine every target is written by the executors — object storage, warehouses over JDBC, and the lakehouse. That last one was the exception until recently: DuckLake has no Spark connector, so a lakehouse target collected the result to the driver and loaded it from there, which meant a pipeline sized for a cluster still had to fit its result in one process.

Two targets still take the collected result, and deliberately: HTTP and SaaS. Each posts a few hundred records at a time to an API, so there is nothing for a cluster to parallelise and the row counts that reach them are small by nature.

Reverse ETL into a SaaS tool#

The SaaS tool target pushes rows back into HubSpot or Salesforce through a connection you have already made — the same one that syncs contacts in syncs them back out, so there is no second copy of the CRM's credential to manage. Pick the connection, the object, and the column that identifies a record: a HubSpot unique property such as email, or a Salesforce External ID field. Every other column is sent as a field, and it is an upsert — the same row twice updates rather than duplicates.

• The batch cap is applied for you. HubSpot takes 100 records per request and Salesforce 200; exceeding it rejects the whole batch, not one record.
• The id column is checked before the first request, because otherwise every record is rejected one batch at a time.
• Only HubSpot and Salesforce can be written to. Stripe, Shopify, Jira, Zendesk and Google Sheets are read-only here — creating a charge or an issue from a nightly pipeline is a different kind of decision.
• A shared connection can be read from but not written to. Pushing records into somebody else's CRM is a bigger step than reading rows out of it, so reverse-ETL targets resolve owner-only.
• Partial batches are not rolled back. One bad record does not block the rest; the run fails with the rejections named, and re-running is safe because an upsert on the same key updates rather than duplicates.

The CRM's host must be on the egress allow-list (api.hubapi.com, or your Salesforce My Domain) under Admin → Developer runtime. A run blocked by the proxy says so by name rather than leaving you with a bare 403.

Where the data goes next#

1. 1

   The run loads tables into your destination

   Files under dataset/table folders in a bucket, or rows in a database table. Every target reports the schema it loaded in the run's metrics.
2. 2

   The destination is re-crawled automatically

   After every successful run the linked catalog source is crawled, so new tables appear as assets without waiting for a crawl schedule.
3. 3

   Streaming sources: Kafka, Kinesis, Pub/Sub

   A Kafka / Redpanda topic, a Kinesis stream or a Pub/Sub subscription is a source node read in micro-batches on the schedule: each run continues from where the last one durably loaded (offsets per partition, sequence numbers per shard, kept as the engine cursor and persisted only when the load committed - at-least-once, never lost), up to a message cap or until the stream goes quiet. Rows carry the payload's fields plus _stream_* metadata; credentials are secrets by name; the broker host must be on the sandbox egress allow-list.
4. 4

   Streamed rows and reverse ETL

   Push JSON rows to /api/etl/ingest under the pipeline's trigger token and an ingest source drains them exactly once per run. On the way out, an HTTP API target sends rows to any external endpoint in authenticated JSON batches.
5. 5

   Change data capture from PostgreSQL

   A database source in CDC mode streams inserts, updates and deletes from a logical-replication slot the engine creates and manages. Point it at a Delta merge target and you get a continuously-applied mirror of the source table — deletes included.
6. 6

   Failures reach you where you work

   Per-pipeline alerts on failure, recovery, or every success — delivered in-app and mirrored to the Slack, Teams, Discord or webhook channels connected on the Integrations page.
7. 7

   Preview any node on sampled data

   Select a node and Preview data runs its upstream steps in the sandbox on sampled sources, showing the rows and column types that node produces — before anything is loaded anywhere.
8. 8

   Every meaningful save is a version

   The Settings tab keeps a version history of the pipeline's graph and code. Restore any version with one click — the restore itself becomes the newest version, so nothing is ever lost.
9. 9

   Schema drift is caught before the load

   Targets can evolve silently (default), warn, or fail on drift. Strict targets compare the incoming frame against last run's stored shape and abort before writing — the error names exactly which columns were added, removed or retyped.
10. 10

    Quality gates keep bad rows out

    A Quality gate transform checks columns mid-pipeline — not null, unique, in range, regex, allowed values, minimum row count. Per rule, a violation fails the run, warns, or drops the offending rows; every outcome lands in the run's quality metrics and [quality] log lines.
11. 11

    Lineage is registered in the catalog

    Each run records which sources fed which produced assets. Open an asset in the Data Catalog and the drawer shows its upstream edges — a storage path, a database table, an API URL, or a Python script.
12. 12

    Everything downstream can use it

    Catalog assets back BI dashboards, AI Analyst questions, agent SQL tools and swarm retrieve nodes — loaded data is queryable the moment the crawl lands.

Observability, logs, audit and IAM#

The Runs tab is the record: status, trigger, attempt count, pinned parameters, duration, rows loaded per target, and the sandbox's output with secret values scrubbed — streamed live while the run executes, so a long job narrates itself instead of going dark until the end. A failed scheduled run also sends a notification — a pipeline that fails silently at 3am is the failure mode the runs table exists to prevent. Pipeline creation, edits, deletion, run starts and outcomes are written to the audit log; runs are server-written rows a client cannot forge or edit. Connections honour IAM: a connection shared with you works in a pipeline exactly as it does in BI, and the AI assist obeys your model allow-lists.

Data-size limits and machine sizing#

Each run executes in one sandbox container as an in-memory process — there is no distributed engine, so a single run never spans machines and its working set must fit in the container's RAM. Rule of thumb: transforms need 3–5× the raw data size in memory (joins, wide aggregations and SCD comparisons sit at the high end). The per-kernel ceiling is the batch memory limit under Admin → Developer runtime (default 4 GB, 2 CPUs).

The host figures cover kernels plus the app; multiply the kernel column by how many runs you allow at once. Past a few GB per run, change the shape of the work instead of the machine: narrow reads with incremental cursors or CDC, load raw into Snowflake / BigQuery / Databricks and transform there (ELT), or split into chained pipelines with bounded working sets. A run that outgrows its kernel dies as a failed run whose logs end abruptly — that is the container OOM.

Horizontal scaling#

Two layers scale independently. Run execution scales out through the runtime backend: docker runs kernels on one host (scale that host up), k8s schedules every run as its own pod across the cluster — the horizontal path — and e2b rents externally hosted sandboxes. The unit of parallelism is the run: ten pipelines can execute on ten nodes, but one run's dataframe lives on one machine — unless the pipeline is on the Spark engine, where the data is spread across the cluster's executors (see Engines above).

The app tier is safe behind a load balancer. Replicas are stateless (all state is in the database) and every scheduler decision is an atomic claim: a due pipeline's clock advance is a compare-and-set one replica wins, a due retry claims retrying→queued with one winner, and finalisation claims the terminal status exactly once — so chains, alerts, lineage and crawls cannot double-fire even when replicas race. No sticky sessions needed; trigger, ingest and result callbacks work on any replica. The one per-host concern is the egress allowlist, which each Docker host's proxy reads locally.

Limits, stated plainly#

• Database connectivity covers the three wire families in the table above; the rest stage through object storage. The refusal happens at save time, with the message telling you so.
• One run = one container: data is processed in memory on a single machine. Size guidance and the scale-out story live in the two sections above.
• The first run pays a cold start plus a package install — a couple of minutes for the full stack. Later runs on a warm image are much faster.
• Three runs may execute concurrently per account; the trigger endpoint answers 409 when the pipeline has no capacity to start.