derivation-aggregate-metrics

Stateless Estuary derivation that continuously aggregates source documents into metrics, using schema-level reduction annotations to combine documents with the same key.

Prereq: read derivation-basics first for concepts, project layout, workflow, and the stateless-vs-stateful distinction.

Docs:

• https://docs.estuary.dev/getting-started/tutorials/continuous-materialized-view/ — end-to-end tutorial with reductions
• https://docs.estuary.dev/reference/reduction-strategies/ — reference for sum, merge, minimize, maximize, append, firstWriteWins, lastWriteWins

When to use this over alternatives

• Daily / hourly / per-customer aggregations: sum orders by day, count events per user, etc.
• Running totals: lifetime revenue per customer that updates in real time
• Statistical metrics per group: min / max / sum / count per sensor, per region
• Continuous materialised views: stream-native replacement for scheduled aggregation queries

Reach for other skills when:

• Output is 1:1 with input (no grouping) → derivation-filter-transform
• Output is many-per-input (not grouping) → derivation-flatten-array
• Combining across multiple source collections on a shared key → derivation-join-collections (which uses the same reduction mechanics)
• Custom state you update procedurally (not pure reduction) → derivation-stateful-logic

How it works (one paragraph)

You don't write aggregation logic yourself. Your lambda emits one "delta" document per source input (1 AS order_count, $total AS revenue). The derived collection's schema carries reduce: annotations telling Estuary's runtime how to merge documents with the same key (sum adds numeric values, maximize keeps the latest timestamp, etc.). At materialisation time, the runtime reduces all deltas with the same key into a single final document. This is why shuffle: any works even though the effect is aggregation — the lambda itself is stateless.

Canonical Example — SQL

Aggregate orders into daily metrics: order count, total revenue, total items, smallest/largest order.

Project layout

my-derivation/
├── flow.yaml
└── schema.yaml

schema.yaml

Top-level reduce: merge is mandatory — it tells Estuary to merge documents with the same key rather than overwrite. Each aggregated field carries its own strategy:

type: object
properties:
  date:
    type: string
    format: date

  # sum-of-ones counts
  order_count:
    type: integer
    default: 0
    reduce: { strategy: sum }

  # sum of per-order values
  total_revenue:
    type: number
    default: 0
    reduce: { strategy: sum }
  total_items:
    type: integer
    default: 0
    reduce: { strategy: sum }

  # min / max tracking
  smallest_order:
    type: number
    reduce: { strategy: minimize }
  largest_order:
    type: number
    reduce: { strategy: maximize }

reduce: { strategy: merge }
required: [date]

default: 0 on numeric sum fields is optional for the reduction itself (sum implicitly treats a missing LHS as 0). Keep it if downstream consumers expect a non-null field on the first delta — otherwise it can be dropped.

flow.yaml

collections:
  acmeCo/analytics/daily-order-metrics:
    schema: schema.yaml
    key: [/date]
    derive:
      using:
        sqlite: {}
      transforms:
        - name: perOrder
          source: acmeCo/production/orders
          shuffle: any
          lambda: |
            SELECT
              date($created_at) AS date,
              1                 AS order_count,
              $total            AS total_revenue,
              $item_count       AS total_items,
              $total            AS smallest_order,
              $total            AS largest_order;

Each source order becomes one tiny "delta" doc. The runtime reduces all deltas with the same date into the final aggregate. shuffle: any is correct — the lambda is per-document with no state; reduction happens downstream.

Canonical Example — TypeScript

TypeScript is rarely needed here — the lambda logic is trivial — but it helps when you need conditional emissions or computed buckets:

import { IDerivation, Document, SourcePerOrder } from 'flow/acmeCo/analytics/daily-order-metrics.ts';

export class Derivation extends IDerivation {
  perOrder(_read: { doc: SourcePerOrder }): Document[] {
    const order = _read.doc;

    return [{
      date: order.created_at.split('T')[0],
      order_count: 1,
      total_revenue: order.total,
      total_items: order.item_count,
      smallest_order: order.total,
      largest_order: order.total,
      // Example of computed bucket only TS can do cleanly:
      // high_value_count: order.total > 1000 ? 1 : 0,
    }];
  }
}

Output schema and flow.yaml are the same — just swap sqlite: {} for the TypeScript module.

Test

Local validation — flowctl preview --fixture

Reductions behave a bit oddly under preview. Preview may show partially-reduced output — Estuary reduces within a transaction boundary but not across, so with 3 same-day orders you might see 1, 2, or 3 docs for that date depending on how the events were batched. The full reduction happens at materialisation time, not at emit time. Don't assert on exact preview doc counts; assert on totals-per-key after materialisation.

fixture.jsonl:

["acmeCo/production/orders", {"order_id": "o1", "created_at": "2024-06-01T09:00:00Z", "total": 49.98,  "item_count": 2}]
["acmeCo/production/orders", {"order_id": "o2", "created_at": "2024-06-01T13:30:00Z", "total": 250.00, "item_count": 5}]
["acmeCo/production/orders", {"order_id": "o3", "created_at": "2024-06-01T18:15:00Z", "total": 12.50,  "item_count": 1}]
["acmeCo/production/orders", {"order_id": "o4", "created_at": "2024-06-02T10:00:00Z", "total": 99.00,  "item_count": 3}]
{"commit": true}

flowctl preview --source flow.yaml \
  --name acmeCo/analytics/daily-order-metrics \
  --fixture fixture.jsonl --timeout 30s

With this single-transaction fixture (all 4 orders before one {"commit": true}), the runtime reduces every same-key doc within the transaction, producing one fully-reduced doc per date:

{"date":"2024-06-01","order_count":3,"total_revenue":312.48,"total_items":8,"smallest_order":12.50,"largest_order":250.00}
{"date":"2024-06-02","order_count":1,"total_revenue":99.00,"total_items":3,"smallest_order":99.00,"largest_order":99.00}

To see partial / pre-reduction deltas (what the materialization receives in production when events arrive in separate transactions), add intermediate {"commit": true} markers between orders in the fixture. Each transaction's deltas are then emitted separately.

Spec-defined tests — tests: block

Note that verify compares against the reduced result — so the test asserts totals, not per-delta docs:

tests:
  acmeCo/tests/daily-order-metrics:
    - ingest:
        collection: acmeCo/production/orders
        documents:
          - { order_id: "o1", created_at: "2024-06-01T09:00:00Z", total: 49.98,  item_count: 2 }
          - { order_id: "o2", created_at: "2024-06-01T13:30:00Z", total: 250.00, item_count: 5 }
          - { order_id: "o3", created_at: "2024-06-01T18:15:00Z", total: 12.50,  item_count: 1 }
          - { order_id: "o4", created_at: "2024-06-02T10:00:00Z", total: 99.00,  item_count: 3 }
    - verify:
        collection: acmeCo/analytics/daily-order-metrics
        documents:
          - { date: "2024-06-01", order_count: 3, total_revenue: 312.48, total_items: 8, smallest_order: 12.50, largest_order: 250.00 }
          - { date: "2024-06-02", order_count: 1, total_revenue: 99.00,  total_items: 3, smallest_order: 99.00,  largest_order: 99.00 }

Variations

• Multi-dimensional grouping: key: [/date, /region, /category]. Extend the lambda to SELECT date($created_at), $region, $category, .... Works the same — reduction happens per composite key.
• Running totals (lifetime per customer): key: [/customer_id], no date bucket. The reduce strategies stay the same.
• Hourly buckets: strftime('%Y-%m-%dT%H:00:00Z', $timestamp) instead of date(). The key becomes the hourly bucket.
• Deriving averages: store sum and count separately, divide at query time. Averages can't be reduced (averaging two averages isn't equivalent).
• "Latest value" for string fields: use reduce: { strategy: lastWriteWins }. (maximize keeps the lexicographically greatest value, not the most-recently-written one — "zzz" arriving before "aaa" would still win.)
• Collecting into arrays: reduce: { strategy: append }. Use sparingly — arrays grow unbounded if the key cardinality × events-per-key is high.

Gotchas specific to aggregate-metrics

• This reduction-based aggregation pattern requires standard-updates mode on the destination. Reduction happens at materialization time when Estuary loads the existing row, reduces incoming deltas into it, and writes back the reduced result — that only happens in standard-updates mode (the default on every connector). If the destination must use delta_updates: true (opt-in on Snowflake / BigQuery / Databricks for cost or latency reasons), Estuary appends each per-transaction delta as-is and your table will show fragments, not running totals. Either reduce downstream in the query layer, or use a stateful aggregation via derivation-stateful-logic — maintain the running total in SQLite and emit the current total per input so no downstream reduction is needed. See Delta updates and the worked sum example in materialization-protocol.
• Preview shows pre-reduction deltas. This is not a bug — the runtime reduces at materialisation time. If you want to see final reduced state locally, materialise to a local Postgres (via the flowctl preview materialization path or docker-compose Postgres in projects/customer-llm-skills/).
• reduce: merge on the top level is mandatory. Without it, Estuary overwrites documents with the same key instead of merging field-by-field. Result: only the last delta survives, no summing.
• Floating-point drift on big sums. For financial data, store amounts as integer cents (total_cents: integer) and format at query time. Summing millions of floats accumulates imprecision.
• Key cardinality blows up state. Don't group by things with unbounded cardinality (e.g., event IDs). Pick grouping dimensions that bound the number of keys.
• Schema changes to reductions may require reset. Changing a field's reduce: strategy after data exists is not always compatible — you may need to publish a new collection and backfill.

Related

• derivation-basics — prerequisite
• derivation-join-collections — uses the same reduction mechanics to merge across multiple sources
• derivation-stateful-logic — when you need procedural state (balances, approval counts), not pure reduction
• derivation-windowing — time-bounded aggregation (last 24h), when simple per-day buckets aren't enough
• Continuous materialised view tutorial
• Reduction strategies reference