A Hands-On Guide to Building Streaming Transformation with Estuary Derivation Agent Skills

Introduction

AI-powered development environments like Claude Code and OpenAI Codex are changing how data engineers build and manage pipelines. Instead of manually writing configuration files, transformation logic, and deployment steps from scratch, engineers can increasingly describe what they want to build in natural language and use AI agents to generate, test, and refine the implementation.

Estuary Agent Skills bring this workflow into real-time ETL/ELT development. They give AI coding assistants the context needed to work with Estuary, so you can create captures, materializations, and streaming Derivations directly from your development environment.

In this guide, we'll use a PostgreSQL banking dataset to build a series of streaming transformations with Estuary Derivation Agent Skills, focusing specifically on Derivation Skills. We'll:

1. Filter approved transactions from the source stream
2. Calculate real-time account-level transaction metrics such as lifetime count, total spend, and minimum and maximum transaction amounts
3. Maintain state to calculate average transaction amounts as new transactions arrive
4. Join transactions with account, merchant, and customer data to create an enriched analytics-ready dataset
5. Flatten nested transaction tags into individually queryable records
6. Track account activity over a rolling 24-hour window and flag unusually high transaction activity
7. Use Python embeddings to classify merchants into predefined categories such as grocery, travel, restaurant, gas, retail, online, and entertainment

The goal is not just to show what each Derivation Skill can generate, but to explain how the resulting configuration, state, and transformation logic work together in a real-time streaming pipeline before the data is materialized into ClickHouse for analytics.

How Do Estuary Agent Skills Work?

Estuary Agent Skills let you build and operate data pipelines directly in your AI coding assistant. Instead of switching between your IDE and the Estuary dashboard, skills teach your assistant to run specific flowctl workflows—like setting up a Postgres capture, materializing into Snowflake, or diagnosing pipeline issues—all without leaving your editor.

These portable skills follow the open-source estuary/agent-skills repository and SKILL.md standard, making them compatible with Claude Code, Cursor, OpenAI Codex, GitHub Copilot, Gemini CLI, and other AI development environments.

What Are Estuary Derivation Agent Skills?

Estuary Derivation Agent Skills help AI coding assistants build in-flight streaming data transformations from natural-language instructions. They can generate the required configuration and transformation logic without requiring you to manually write every SQL query, configuration file, or custom ETL job.

Instead of manually building a derivation, describe it in plain language:

"Create a derivation that joins banking transactions with customer and merchant information, filters fraudulent transactions, calculates a running account balance, and outputs an enriched collection ready for ClickHouse analytics."

The agent will generate the configuration, guide you through deployment, integrate the transformed data into your pipeline, and handle schema validation—all automatically within your IDE.

Prerequisites

To follow along, you'll need:

• An Estuary account
• The flowctl CLI installed and authenticated (see the flowctl installation guide).
• A supported AI coding assistant: Claude Code, Cursor, OpenAI Codex, or another compatible tool
• A source collection (I'll use a PostgreSQL banking dataset in this example, but any transactional dataset will work)
• A materialization destination: Clickhouse to inspect the transformed data
• An AI assistant installed locally with access to the Estuary Agent Skills (I'll use Claude throughout this article, though the same skills can be used with other supported AI assistants like Open AI)
• Docker set up and ngrok running in the background, to host a local Postgres instance and expose it for Estuary's capture connector to reach. You will also need it to run Python Derivations.

Source Code: The complete code for this walkthrough is available in the accompanying GitHub repository. It includes the Estuary capture and materialization configurations, along with the flow.yaml, schema.yaml, SQL, and Python files generated for each derivation example. Each transformation is organized in its own directory so you can follow along with the examples or reuse individual patterns in your own project.

Installing Estuary Agent Skills

Estuary's Agent Skills cover several operations, including Captures, Materializations, and Derivations.

The Agent Skills can be installed as plugins from the marketplace in Claude Code:

/plugin marketplace add estuary/agent-skills

Then install the plugins you want:

/plugin install estuary-captures@estuary

/plugin install estuary-materializations@estuary

/plugin install estuary-derivations@estuary

After installing each plugin, refresh it with /reload-plugin. To invoke the skills, you can write a prompt such as: "Capture from my Postgres and materialize into ClickHouse."

What Streaming Transformation Patterns Can Estuary Agent Skills Build?

After installing the derivation skills, here are the skills you can work with:

1. derivation-basics: Introduces foundational concepts for Estuary derivations — covers what they are, when not to use one, language choices (SQLite/TypeScript/Python), schema setup, stateless vs. stateful design, the workflow, and testing.
2. derivation-filter-transform: Filtering and transforming records is one of the most common steps in any data pipeline, whether you're cleaning data, removing unwanted events, or standardizing fields before analytics. Use this derivation when you need to reshape incoming records in real time without modifying the original source collection.
3. derivation-aggregate-metrics: Aggregation derivations are useful for maintaining continuously updated metrics as new events arrive, eliminating the need for scheduled batch jobs. They're commonly used to power real-time dashboards, KPI reporting, operational monitoring, and customer or account-level summaries.
4. derivation-flatten-array: Many APIs and transactional systems store nested JSON arrays that are difficult to analyze in analytical databases. This derivation is useful for converting arrays into individual records, making it easier to filter, aggregate, and visualize each element independently.
5. derivation-join-collections: Data often lives across multiple tables, making joins essential for building analytics-ready datasets. Use this derivation to enrich streaming events with related information from other collections, creating denormalized records that are easier to query downstream.
6. derivation-stateful-logic: Some business rules require remembering previous events rather than evaluating each record in isolation. Use stateful derivations for streaming use cases such as running balances, fraud detection, inventory tracking, approval workflows, deduplication, or any logic that depends on historical state.
7. derivation-windowing: Maintains a stateful, time-bounded sliding window of recent events (e.g., "failed logins in the last 5 minutes" or "transfers in the last 24h") using SQLite state plus readDelay to expire old entries. This pattern is common for fraud detection, rate limiting, and rolling-window alerts.
8. derivation-python: Python derivations are useful when transformations go beyond SQL or SQLite and require external libraries or services. Use them for machine learning inference, embedding generation, asynchronous API calls, custom validation, or other advanced business logic that benefits from Python's ecosystem. They follow the same general shape as TypeScript derivations but use Python with Pydantic types. They run only on private/BYOC data planes, not shared infrastructure, so confirm that your setup supports them before using this skill.

Streaming Transformation Architecture

The pipeline captures PostgreSQL banking data into Estuary collections, applies streaming transformations through derived collections, and materializes the transformed results into ClickHouse. The walkthrough progressively adds filtering, aggregation, stateful processing, joins, array flattening, windowing, and Python-based inference.

What the agent does vs. what you review

Agent Skills can automate much of the mechanical work involved in creating a derivation: selecting a transformation pattern, generating flow.yaml and schema definitions, creating SQL or Python logic, previewing results, and guiding deployment. However, the generated configuration should still be reviewed like any other production code. In particular, verify collection keys, shuffle behavior, schema constraints, state-growth characteristics, retention logic, and the generated SQL or Python before publishing the derivation.

Walkthrough of the Estuary Derivation Agent Skills

Step 1: Capture PostgreSQL banking data

Assuming Docker and ngrok are already running in the background, you can prompt Claude with "Capture from my Postgres," which triggers estuary-captures:capture-postgres-create. From there, select self-hosted and ngrok (local dev) as your connection method, then specify the tables you want to capture — in this example, transactions, customers, merchants, and accounts.

Claude will then ask for permission to fetch this content. Provide a capture name, such as demo/banking/source-postgres, and Claude will fetch the connection details needed to run it.

Step 2: Transform the Captured Collections using Derivations

There are a few ways to transform:

Derived Collection

When building a derived collection, the central question is where accumulation will happen: within the derivation state, or within an external database that you materialize into? Both approaches can produce equivalent results, but they do it in very different ways.

In these examples, the AI agent creates the derivation using SQLite by default. The derived collection has its own JSON schema, key, and specification file (typically a nested flow.yaml). Every collection has an associated schema (schema.yaml) that documents must validate against, whether the collection is captured directly or derived from other collections. The schema is enforced on writes and reads, and Estuary pauses the task if a document does not match it rather than allowing invalid data to flow downstream.

Let’s try a few examples using the Derivation Agent Skills:

1. Filter Transform

Let's start with a simple transformation. In our banking dataset, transactions can have a status of approved, pending, or declined, but for many downstream analytics we only want successfully completed transactions. We'll use the Filter Transform skill to create a new derived collection containing only approved transactions while leaving the original source collection unchanged.

In Claude, run: derivation-filter-transform

The agent asks what you want the derivation to do to the data, and offers a few starting points — filter rows by a condition, compute or cleanse fields, or both:

For this example, we'll filter rows by a condition, keeping only transactions where the status is approved:

The agent doesn't just generate a filter and stop there—it previews the derivation against your real Postgres data to confirm that the logic holds. In this case, it verified that the source collection contains all three statuses (approved, declined, and pending) rather than assuming the filter worked simply because every previewed row happened to be approved.

Output:

This generates two local files under your estuary-derivation project folder:

• flow.yaml — defines a new derived collection, banking/derived/approved-transactions, sourced from banking/public/transactions, and keeps only rows where status = 'approved':

flow.yaml:

yaml language-plaintext
collections:
RuheeShrestha/banking/derived/approved-transactions:
  writeSchema: schema.yaml
  readSchema:
    allOf:
    - $ref: flow://write-schema
    - $ref: flow://inferred-schema
  key:
  - /transaction_id
  derive:
    using:
      sqlite: {}
    transforms:
    - name: filterApprovedTransactions
      source: RuheeShrestha/banking/public/transactions
      shuffle: any
      lambda: |
        SELECT JSON($flow_document) WHERE $_meta$op != 'd' AND $status = 'approved';
        SELECT $transaction_id, $_meta WHERE $_meta$op = 'd';
        SELECT $transaction_id, $_meta WHERE $_meta$op = 'd';

The lambda only forwards non-delete rows when status = 'approved' and explicitly handles _meta.op = 'd'; it has no branch for an update that changes status from matching to non-matching.

CDC filtering caveat: This transformation correctly removes records when the source transaction itself is deleted, but it does not remove records when the filtered field changes. For example, if a transaction changes from approved to declined, it no longer passes the filter, but the previously written approved record remains in the derived collection. This is a common issue when filtering CDC streams on mutable columns, because updates that stop matching the filter need to explicitly produce a deletion for the derived record.

schema.yaml — a minimal write schema, since transaction_id is the only field the derivation itself needs to enforce as a key:

yaml language-plaintext
type: object
properties: transaction_id:   type: string
   format: uuid
required: [transaction_id]

Materializing it

Once published, the derived collection is another source you can point a materialization at, such as Postgres, Snowflake, or ClickHouse. Using the materialize-clickhouse-create skill, we can point the collection to ClickHouse, which is added as a binding in our estuary-materialization project's flow.yaml:

And in ClickHouse, the filtered result is queryable like any other table:

sql language-plaintext
SELECT * FROM approved_transactions;

2. Aggregate Metrics

Now that we've filtered the data, we can build real-time metrics on top of the approved transactions. Rather than recalculating summaries in a downstream database, we'll use an aggregation derivation to continuously maintain per-account metrics such as transaction count, total spend, and the smallest and largest transaction amounts.

In Claude, run: derivation-aggregate-metrics

The agent asks what you want to aggregate and by what grouping:

We can prompt something like:

“Aggregate the transactions table by account_id, calculating total amount, minimum amount, and maximum amount for each account, as well as the lifetime transaction count for approved transactions only.”

This creates a flow.yaml file and a schema.yaml file within the estuary-derivation/account-transaction-metric directory.

flow.yaml file:

yaml language-plaintext
collections:
 RuheeShrestha/banking/derived/account-transaction-metrics:
   schema: schema.yaml
   key: [/account_id]
   derive:
     using:
       sqlite: {}
     transforms:
       - name: perApprovedTransaction
         source: RuheeShrestha/banking/derived/approved-transactions
         shuffle: any
         lambda: |
           SELECT
             $account_id AS account_id,
             1 AS transaction_count,
             CAST(ROUND(CAST($amount AS REAL) * 100) AS INTEGER) AS total_amount_cents,
             CAST(ROUND(CAST($amount AS REAL) * 100) AS INTEGER) AS min_amount_cents,
             CAST(ROUND(CAST($amount AS REAL) * 100) AS INTEGER) AS max_amount_cents
           WHERE $_meta$op != 'd' AND $account_id IS NOT NULL;

The generated flow.yaml uses the same expression for both the minimum and maximum transaction amounts.The following schema.yaml explains the reason for it being because of the reduce function being implemented:

schema.yaml file:

yaml language-plaintext
type: object
properties:
 account_id:
   type: string
   format: uuid
 transaction_count:
   type: integer
   default: 0
   reduce: { strategy: sum }
 total_amount_cents:
   type: integer
   default: 0
   reduce: { strategy: sum } min_amount_cents:
   type: integer
   reduce: { strategy: minimize }
 max_amount_cents:
   type: integer
   reduce: { strategy: maximize }
reduce: { strategy: merge }
required: [account_id]

CDC aggregation assumption: This aggregation assumes that approved transactions are insert-only. If an existing PostgreSQL transaction is updated, the new CDC event is treated as another contribution to transaction_count and total_amount_cents; the sum reducer does not automatically retract the previous value, so updates can overstate the resulting metrics.

Materializing it

Using the same pattern as before, we can add /banking/derived/account-transaction-metrics as a binding in your materialization's flow.yaml, pointing to an account_transaction_metrics table in ClickHouse.

3. Stateful Logic

Some streaming transformations require remembering what happened previously rather than processing each event independently. In this example, we'll extend the previous example, which did not include the requested average transaction amount.

Since the average depends on the aggregated total_amount_cents and transaction_count, it would need to be calculated from those lifetime metrics, or in an additional derivation that requires stateful logic.

Since this requires stateful logic, the agent pulls the stateful-logic derivation skill to do this, and creates a new average-transaction derivation. This creates 4 new files: init.sql, lambda.sql, flow.yaml and schema.yaml.

init.sql:

sql language-plaintext
CREATE TABLE IF NOT EXISTS account_totals (
 account_id TEXT PRIMARY KEY NOT NULL,
 transaction_count INTEGER NOT NULL DEFAULT 0,
 total_amount_cents INTEGER NOT NULL DEFAULT 0
);

lambda.sql contains the transformation logic. The lambda UPSERTs into an internal SQLite table (account_totals), incrementing transaction_count and total_amount_cents for that account.

sql language-plaintext
INSERT INTO account_totals (account_id, transaction_count, total_amount_cents)
SELECT $account_id, 1, CAST(ROUND(CAST($amount AS REAL) * 100) AS INTEGER)
WHERE $_meta$op != 'd' AND $account_id IS NOT NULLON CONFLICT(account_id) DO UPDATE SET
 transaction_count = transaction_count + 1,
 total_amount_cents = total_amount_cents + excluded.total_amount_cents;
SELECT
 account_id,
 transaction_count,
 total_amount_cents,
 CAST(total_amount_cents AS REAL) / transaction_count AS average_amount_cents
FROM account_totals
WHERE account_id = $account_id AND $_meta$op != 'd' AND $account_id IS NOT NULL;

Insert-Only Assumption for Stateful Averages: Like the previous aggregation example, this stateful average assumes transactions are insert-only; an update to an existing transaction would increment the stored count and total again rather than replacing the transaction's previous contribution.

flow.yaml references init.sql for state initialization and lambda.sql for the transformation logic:

yaml language-plaintext
collections:
 RuheeShrestha/banking/derived/account-average-transaction:
   schema: schema.yaml
   key: [/account_id]
   derive:
     using:
       sqlite:
         migrations:
           - init.sql
     transforms:
       - name: perApprovedTransaction
         source: RuheeShrestha/banking/derived/approved-transactions
         shuffle: { key: [/account_id] }
         lambda: lambda.sql

The expected output consists of transaction_count, total_amount_cents, and average_amount_cents, as defined in the schema.yaml.

yaml language-plaintext
type: object
properties:  account_id:    type: string
    format: uuid
  transaction_count:    type: integer
  total_amount_cents:    type: integer
  average_amount_cents:    type: number
required: [account_id]

For each approved transaction, a keyed shuffle (shuffle: { key: [/account_id] }) routes every event for the same account to the same shard, so they share state. It then re-reads the updated row and emits the current average (total_amount_cents / transaction_count) as the output document.

4. Join Collections

Transaction records alone don't provide much business context in your data warehouse. To make them analytics-ready, we'll enrich each transaction with information from the associated account, merchant, and customer collections, producing a single denormalized dataset that is easier to query in ClickHouse.

In Claude, run: derivation-join-collections

The Claude agent loads the skill. We can prompt it with something like:

“Create an inner join across transactions, accounts, merchants, and customers.”

As a result, each pairwise join creates transactions-joined with a flow.yaml and schema.yaml.

flow.yaml file consists of its initialization of account_dim, merchant_dim, customer_dim, pending_transactions table:

sql language-plaintext
CREATE TABLE account_dim (
              account_id TEXT NOT NULL PRIMARY KEY,
              customer_id TEXT,
              account_type TEXT,
              account_status TEXT,
              account_balance TEXT
            );
            CREATE TABLE merchant_dim (
              merchant_id TEXT NOT NULL PRIMARY KEY,
              merchant_name TEXT,
              merchant_category TEXT,
              merchant_city TEXT,
              merchant_state TEXT
            );
            CREATE TABLE customer_dim (
              customer_id TEXT NOT NULL PRIMARY KEY,
              customer_name TEXT,
              customer_email TEXT,
              customer_city TEXT,
              customer_state TEXT
            );
            CREATE TABLE pending_transactions (
              transaction_id TEXT NOT NULL PRIMARY KEY,
              account_id TEXT NOT NULL,
              merchant_id TEXT NOT NULL,
              doc JSON NOT NULL
            );

..and its transformations:

yaml language-plaintext
transforms:
      - name: fromAccounts
        source: RuheeShrestha/banking/public/accounts
        shuffle: { key: [/account_id] }
        lambda: |
          INSERT INTO account_dim (account_id, customer_id, account_type, account_status, account_balance)
          SELECT $account_id, $customer_id, $account_type, $status, $balance
          WHERE $_meta$op != 'd'ON CONFLICT (account_id) DO UPDATE SET
            customer_id = excluded.customer_id,
            account_type = excluded.account_type,
            account_status = excluded.account_status,
            account_balance = excluded.account_balance;

          SELECT JSON_PATCH(JSON_PATCH(JSON_PATCH(p.doc,
              JSON_OBJECT('account_type', a.account_type, 'account_status', a.account_status, 'account_balance', a.account_balance)),
              JSON_OBJECT('merchant_name', m.merchant_name, 'merchant_category', m.merchant_category, 'merchant_city', m.merchant_city, 'merchant_state', m.merchant_state)),
              JSON_OBJECT('customer_id', a.customer_id, 'customer_name', c.customer_name, 'customer_email', c.customer_email, 'customer_city', c.customer_city, 'customer_state', c.customer_state)
            ) AS json
          FROM pending_transactions p
          JOIN account_dim a ON a.account_id = p.account_id
          JOIN merchant_dim m ON m.merchant_id = p.merchant_id
          JOIN customer_dim c ON c.customer_id = a.customer_id
          WHERE p.account_id = $account_id;

      - name: fromMerchants
        source: RuheeShrestha/banking/public/merchants
        shuffle: { key: [/merchant_id] }
        lambda: |
          INSERT INTO merchant_dim (merchant_id, merchant_name, merchant_category, merchant_city, merchant_state)
          SELECT $merchant_id, $merchant_name, $category, $city, $state
          WHERE $_meta$op != 'd'ON CONFLICT (merchant_id) DO UPDATE SET
            merchant_name = excluded.merchant_name,
            merchant_category = excluded.merchant_category,
            merchant_city = excluded.merchant_city,
            merchant_state = excluded.merchant_state;

          SELECT JSON_PATCH(JSON_PATCH(JSON_PATCH(p.doc,
              JSON_OBJECT('account_type', a.account_type, 'account_status', a.account_status, 'account_balance', a.account_balance)),
              JSON_OBJECT('merchant_name', m.merchant_name, 'merchant_category', m.merchant_category, 'merchant_city', m.merchant_city, 'merchant_state', m.merchant_state)),
              JSON_OBJECT('customer_id', a.customer_id, 'customer_name', c.customer_name, 'customer_email', c.customer_email, 'customer_city', c.customer_city, 'customer_state', c.customer_state)
            ) AS json
          FROM pending_transactions p
          JOIN account_dim a ON a.account_id = p.account_id
          JOIN merchant_dim m ON m.merchant_id = p.merchant_id
          JOIN customer_dim c ON c.customer_id = a.customer_id
          WHERE p.merchant_id = $merchant_id;

      - name: fromCustomers
        source: RuheeShrestha/banking/public/customers
        shuffle: { key: [/customer_id] }
        lambda: |
          INSERT INTO customer_dim (customer_id, customer_name, customer_email, customer_city, customer_state)
          SELECT $customer_id, $name, $email, $city, $state
          WHERE $_meta$op != 'd'ON CONFLICT (customer_id) DO UPDATE SET
            customer_name = excluded.customer_name,
            customer_email = excluded.customer_email,
            customer_city = excluded.customer_city,
            customer_state = excluded.customer_state;

          SELECT JSON_PATCH(JSON_PATCH(JSON_PATCH(p.doc,
              JSON_OBJECT('account_type', a.account_type, 'account_status', a.account_status, 'account_balance', a.account_balance)),
              JSON_OBJECT('merchant_name', m.merchant_name, 'merchant_category', m.merchant_category, 'merchant_city', m.merchant_city, 'merchant_state', m.merchant_state)),
              JSON_OBJECT('customer_id', a.customer_id, 'customer_name', c.customer_name, 'customer_email', c.customer_email, 'customer_city', c.customer_city, 'customer_state', c.customer_state)
            ) AS json
          FROM pending_transactions p
          JOIN account_dim a ON a.account_id = p.account_id
          JOIN merchant_dim m ON m.merchant_id = p.merchant_id
          JOIN customer_dim c ON c.customer_id = a.customer_id
          WHERE c.customer_id = $customer_id;

      - name: fromTransactions
        source: RuheeShrestha/banking/public/transactions
        shuffle: { key: [/account_id] }
        lambda: |
          INSERT INTO pending_transactions (transaction_id, account_id, merchant_id, doc)
          SELECT $transaction_id, $account_id, $merchant_id, JSON($flow_document)
          WHERE $_meta$op != 'd'ON CONFLICT (transaction_id) DO UPDATE SET
            account_id = excluded.account_id,
            merchant_id = excluded.merchant_id,
            doc = excluded.doc;

          SELECT JSON_PATCH(JSON_PATCH(JSON_PATCH(p.doc,
              JSON_OBJECT('account_type', a.account_type, 'account_status', a.account_status, 'account_balance', a.account_balance)),
              JSON_OBJECT('merchant_name', m.merchant_name, 'merchant_category', m.merchant_category, 'merchant_city', m.merchant_city, 'merchant_state', m.merchant_state)),
              JSON_OBJECT('customer_id', a.customer_id, 'customer_name', c.customer_name, 'customer_email', c.customer_email, 'customer_city', c.customer_city, 'customer_state', c.customer_state)
            ) AS json
          FROM pending_transactions p
          JOIN account_dim a ON a.account_id = p.account_id
          JOIN merchant_dim m ON m.merchant_id = p.merchant_id
          JOIN customer_dim c ON c.customer_id = a.customer_id
          WHERE p.transaction_id = $transaction_id;

This code creates one derived collection, transactions-joined, that behaves like a stateful inner join across four streaming collections. Each transform updates local SQLite state and retries pending joins as new records arrive.

Production consideration: This implementation can create an unbounded state because records remain in pending_transactions unless cleanup rules remove them. Even after a transaction successfully joins with the related account, merchant, and customer records and is emitted to transactions-joined, its row remains in SQLite.

To bound the state, delete a pending transaction after a successful join: DELETE FROM pending_transactions WHERE transaction_id = $transaction_id or expire unmatched records after a defined retention period.

Multi-shard consideration: Because the transforms use different shuffle keys (/account_id, /merchant_id, and /customer_id), this pattern is safest as a single-shard example. In multi-shard deployments, related state may be split across different SQLite databases, so production joins should be chained pairwise on a shared key.

Materialization

In ClickHouse, this materializes to banking.transactions-joined, with each output row identified by transaction_id.

5. Flatten Array

This skill is useful for converting a nested array or list inside a JSON document into individual rows—one row per array element instead of one row per document.

Our transactions also contain a nested tags array that categorizes each purchase. Many analytical databases work best with one value per row, so we'll flatten this array into individual records, allowing us to analyze tags independently for reporting and filtering.

When we run derivation-flatten-array on claude, the AI agent will flatten the source transactions table and create the following flow.yaml file:

yaml language-plaintext
key: [/transaction_id, /tag]
derive:
 using:
   sqlite: {}
 transforms:
   - name: flattenTags
     source: RuheeShrestha/banking/public/transactions
     shuffle: any
     backfill: 1
     lambda: |SELECT json_set(json_remove($flow_document, '$.tags'),'$.tag', json_each.value)FROM json_each($tags)WHERE $_meta$op != 'd';

The key detail is the composite key, [/transaction_id, /tag]—because one transaction now produces multiple output rows, the key must include the exploded value to keep each row unique. The lambda uses SQLite's json_each() to iterate the array, json_remove() to drop the original tags field, and json_set() to attach the single tag value in its place, so each output document looks like the original transaction without the array and with one tag.

Materialization

So the 5,378 source transactions become 40,670 rows — one per (transaction, tag) pair, and when materializing into ClickHouse:

6. Windowing

Stateful logic vs. windowing: The previous examples retain state for as long as the application requires it. Windowing introduces an explicit time boundary: records enter the state, contribute to rolling calculations for a defined period, and then expire. This makes windowing especially useful when recent activity matters more than lifetime history.

We'll build a rolling transaction window to detect unusual bursts of activity, a common pattern in fraud detection and operational monitoring.

Prompt:

“For each account, maintain a rolling 24-hour window of transactions. Calculate the transaction count and total transaction amount within the window, and flag accounts with more than 20 transactions as unusual activity.”

This generates the state-management SQL files init.sql, add.sql, and remove.sql, along with flow.yaml and schema.yaml.

Init.sql

This creates a new window_transactions table to store records within the rolling 24-hour transaction window.

sql language-plaintext
CREATE TABLE IF NOT EXISTS window_transactions (
transaction_id TEXT PRIMARY KEY NOT NULL,
account_id TEXT NOT NULL,
amount_cents INTEGER NOT NULL,
transaction_time TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_window_account ON window_transactions (account_id);

add.sql

The generated add.sql calculates rolling activity metrics with a CTE that counts the account's current transactions and sums their amounts. The json_object() then builds the output record for the incoming transaction. Because the incoming transaction has not yet been inserted into the state table, the query adds 1 to stats.cnt and adds the incoming transaction amount to stats.total_cents to produce window_transaction_count, window_total_amount_cents, and unusual_activity.

sql language-plaintext
WITH stats AS (
 SELECTCOUNT(*)                       AS cnt,
   COALESCE(SUM(amount_cents), 0) AS total_cents
 FROM window_transactions
 WHERE account_id = $account_id
)
SELECT json_object(
 'transaction_id', $transaction_id,
 'account_id', $account_id,
 'amount_cents', CAST(ROUND(CAST($amount AS REAL) * 100) AS INTEGER),
 'transaction_time', $transaction_time,
 'window_transaction_count', stats.cnt + 1,
 'window_total_amount_cents', stats.total_cents + CAST(ROUND(CAST($amount AS REAL) * 100) AS INTEGER),
 'unusual_activity', json(CASE WHEN stats.cnt + 1 > 20 THEN 'true' ELSE 'false' END)
) AS json_result
FROM stats
WHERE $_meta$op != 'd' AND $account_id IS NOT NULL;
INSERT OR IGNORE INTO window_transactions (transaction_id, account_id, amount_cents, transaction_time)
SELECT $transaction_id, $account_id, CAST(ROUND(CAST($amount AS REAL) * 100) AS INTEGER), $transaction_time
WHERE $_meta$op != 'd' AND $account_id IS NOT NULL;

remove.sql

As the rolling window advances, transactions older than the configured window—for example, 24 hours—are deleted from this table to keep the state bounded.

sql language-plaintext
DELETE FROM window_transactions WHERE transaction_id = $transaction_id;

flow.yaml

yaml language-plaintext
collections:
RuheeShrestha/banking/derived/account-velocity-window:
   writeSchema: schema.yaml
   readSchema:     allOf:
      - $ref: flow://write-schema
      - $ref: flow://inferred-schema   key: [/transaction_id]
   derive:     using:       sqlite:         migrations:
          - init.sql
     transforms:
      - name: addToWindow
         source: RuheeShrestha/banking/public/transactions
         shuffle: { key: [/account_id] }
         lambda: add.sql
      - name: removeFromWindow
         source: RuheeShrestha/banking/public/transactions
         shuffle: { key: [/account_id] }
         readDelay: 24h
         lambda: remove.sql

In flow.yaml, both transforms read the same public/transactions source and are shuffled on /account_id so a given account's state stays on one shard:

• addToWindow (no delay) — on every new transaction, queries the current window (COUNT/SUM from the window_transactions state table) before inserting, emits the enriched doc, then inserts the row.
• removeFromWindow (readDelay: 24h) — processes the same event again 24 hours after it was first read and deletes it from state. The net effect is that each transaction remains in the window for exactly 24 hours before aging out.

schema.yaml:

yaml language-plaintext
type: object
properties:  transaction_id:    type: string
    format: uuid
  account_id:    type: string
    format: uuid
  amount_cents:    type: integer
  transaction_time:    type: string
    format: date-time
  window_transaction_count:    description: Number of transactions for this account in the trailing 24h (including this one)
    type: integer
  window_total_amount_cents:    description: Sum of transaction amounts for this account in the trailing 24h (including this one)
    type: integer
  unusual_activity:    description: "true when window_transaction_count exceeds 20"    type: boolean
required: [transaction_id, account_id]

It generates /derived/account-velocity-window, which includes window_transaction_count for the trailing 24-hour window, window_total_amount_cents for the sum of amounts in that same window, and an unusual_activity flag when the transaction count exceeds 20.

Output:

Python Derivations

This skill comes in handy when SQL is not well suited to the task—for example, asynchronous API calls, ML inference, or embeddings. In this example, we'll use a Python derivation to generate text embeddings from merchant names and transaction descriptions, enabling semantic search, merchant similarity analysis, and AI-powered recommendations downstream.

In Claude, run: derivation-python

We can prompt something like:

“Classify merchant names into categories (grocery, travel, restaurant, gas, retail, online, entertainment) using nearest-category embedding similarity matching the taxonomy already used in banking.merchants.”

Output

The agent picked model2vec — a lightweight embedding model chosen specifically because it has prebuilt wheels that install in the sandboxed container (heavier transformer models don't, per the skill's own constraints). It creates:

• flow.yaml — a Python-based derivation with model2vec + numpy dependencies

 RuheeShrestha/banking/derived/merchant-category-predictions:

yaml language-plaintext
derive:
using:
python:
module: classify_merchant.flow.py
          dependencies:            model2vec: ">=0.3.0"            numpy: ">=2.0"      transforms:
       - name: classifyMerchant
          source: RuheeShrestha/banking/public/merchants
          shuffle: any

It also creates a classify_merchant.flow.py file that loads the model and embeds each category as a short descriptive phrase. For each merchant, it then computes a cosine-similarity match with softmax-scaled confidence:

classify_merchant.flow.py:

python language-plaintext
from collections.abc import AsyncIterator
import numpy as np
import numpy.typing as npt
from model2vec import StaticModel
from RuheeShrestha.banking.derived.merchant_category_predictions import (
  Document,
  IDerivation,
  Request,
)
MODEL_NAME = "minishlab/potion-base-8M"# Standardized taxonomy mirrors banking.merchants.category so predictions are# directly comparable to the source-of-truth values already in Postgres.# Each label is expanded into a short descriptive phrase so the embedding# captures more signal than a single bare word would.CATEGORY_LABELS: dict[str, str] = {
  "grocery": "grocery store or supermarket",
  "travel": "airline, hotel, or travel agency",
  "restaurant": "restaurant, cafe, or dining",
  "gas": "gas station or fuel",
  "retail": "retail or department store",
  "online": "online or e-commerce marketplace",
  "entertainment": "entertainment, movies, or events venue",
}
class Derivation(IDerivation):
  def init(self, open: Request.Open):
      super().init(open)
      self.model = StaticModel.from_pretrained(MODEL_NAME)
labels = list(CATEGORY_LABELS.keys())
      phrases = list(CATEGORY_LABELS.values())
      embeddings: npt.NDArray[np.float64] = np.asarray(self.model.encode(phrases))
      norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
self.category_labels = labels
      self.category_embeddings: npt.NDArray[np.float64] = embeddings / np.clip(
          norms, 1e-9, None
      )
async def classify_merchant(
      self, read: Request.ReadClassifyMerchant
  ) -> AsyncIterator[Document]:
      src = read.doc
      name = src.merchant_name or ""
embedding: npt.NDArray[np.float64] = np.asarray(self.model.encode([name]))[0]
      norm = np.linalg.norm(embedding)
      if norm > 0:
          embedding = embedding / norm
similarities = self.category_embeddings @ embedding
      scaled = similarities * 10.0
      exp = np.exp(scaled - scaled.max())
      probs = exp / exp.sum()
best = int(np.argmax(probs))
  yield Document(
          merchant_id=src.merchant_id,
          merchant_name=name,
          predicted_category=self.category_labels[best],
          confidence=round(float(probs[best]), 4),
      )

We can view the result via flowctl preview against a small fixture (since local Docker preview wasn't available in this environment):

plaintext
merchant-category-predictions preview output
Generated via: flowctl preview --source flow.yaml --name RuheeShrestha/banking/derived/merchant-category-predictions --fixture fixture.jsonl

merchant_name          predicted_category   confidence
Whole Foods Market     grocery              0.6737
Delta Air Lines        travel               0.7135
Olive Garden           restaurant           0.6257
Shell Gas Station      gas                  0.9871
Best Buy               online               0.3767
Amazon.com             online               0.5495
AMC Theatres           entertainment        0.8362

You can read more about transformation using Python here.

Summarizing Derivation Patterns demonstrated:

Wrapping up

In this article, we explored how Estuary's Derivation Agent Skills simplify streaming data transformations. Starting with a PostgreSQL banking dataset, we used AI-assisted workflows to create common transformation patterns—including filtering records, joining collections, aggregating data, flattening nested JSON, windowing, Python-based inference, and stateful account-level metrics—before materializing the transformed data into ClickHouse for analytics.

While each derivation addresses a specific transformation need, the bigger takeaway is the shift in how these pipelines are developed and maintained. Rather than manually writing derivation specifications, searching through documentation, or implementing custom processing logic, you can describe the desired transformation in plain language and have your AI assistant generate Estuary derivations that fit into your existing Flow pipeline. However, the Agent Skills don't remove the need to understand your data pipeline. You still need to review the schema, state behavior, and transformation logic before deployment. For data engineers, that makes AI useful not just for generating snippets of code, but for accelerating complete streaming-data workflows.

This approach offers several advantages:

• Built Where You Already Work: Your team works in the IDE, so the pipeline should be built there too. Estuary Agent Skills install directly into Claude Code, Cursor, and other supported development environments, reducing the need to switch contexts to find column names or connector details.
• Runs on the Estuary Derivation Runtime: Agent-generated derivations use the same Estuary derivation framework as manually authored transformations. The agent changes how the configuration is created; the resulting derivation is still inspectable, versionable, and reviewable before deployment.
• Faster Development and Maintenance: Natural-language prompts can accelerate common transformation tasks and make it easier to update pipeline logic as requirements change.
• Self-Operating, Not Self-Driving: The agent can automate repetitive work, surface issues, and propose changes while engineers remain responsible for reviewing, approving, and managing production behavior.

As AI-assisted development becomes part of modern data engineering, Estuary Agent Skills provide a practical way to bring natural-language workflows into production streaming pipelines. Engineers can move faster while keeping the resulting transformations transparent, reviewable, and under their control.