derivation-aggregate-metrics git:20260707.6b608bfA

Immutable. This exact content is served forever at /api/v1/blob/4e1c48a93bf83406.

---
name: derivation-aggregate-metrics
description: Create an Estuary derivation that continuously aggregates metrics (daily totals, running counts, min/max, lifetime revenue) using reduction annotations. Use for real-time dashboards, continuous materialised views, and replacing batch ETL aggregations with streaming. Derivations add complexity and cost — confirm the user wants a derivation before reaching for this. Use when user says "aggregation derivation", "derivation to sum daily totals", "running-count derivation", "transformation for min/max per group", "derive lifetime metrics", "aggregate-by-date derivation", or "real-time-dashboard derivation".
---

# 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:

```yaml
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`

```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:

```typescript
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`:

```json
["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}
```

```bash
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:

```json
{"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:

```yaml
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](https://docs.estuary.dev/concepts/materialization/#delta-updates) and the worked `sum` example in [materialization-protocol](https://docs.estuary.dev/reference/Connectors/materialization-protocol/#push-only-endpoints--delta-updates).
- **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](https://docs.estuary.dev/getting-started/tutorials/continuous-materialized-view/)
- [Reduction strategies reference](https://docs.estuary.dev/reference/reduction-strategies/)