Groups & windows — many rows into one

The question for this lesson: every node so far has taken one row in and emitted one row out. But often you want a summary: total sales, a count per status, clicks per hour. That means turning many rows into one. How do you do that?

You use an aggregate node, and you tell it how to group the rows.

What you’ll be able to do

• Use an aggregate node with group_by to collapse many rows into a summary.
• Pick the right grouping: one summary overall, or one row per group.
• Bucket time-stamped rows into a tumbling time window.

An aggregate collapses rows into a summary

A transform reshapes each row on its own. An aggregate is different: it gathers many rows into a group and emits one row per group, using aggregate functions like count(*), sum(...), avg(...), min(...), and max(...).

The group_by setting decides what a “group” is:

• group_by: []: an empty list means one group: everything. You get a single summary row.
• group_by: [status]: one group per distinct status. You get one row for each.

- type: aggregate
  name: summary
  input: customers
  config:
    group_by: []                       # one summary over all rows
    cxl: |
      emit customers = count(*)        # how many rows came in
      emit total_value = sum(lifetime_value.to_int())

🌱 New here? — count(*) and sum(...)

count(*) counts the rows in the group. sum(lifetime_value.to_int()) adds up a number across the group, with .to_int() first because lifetime_value is a declared string (the same text-to-number step from lesson 02). avg, min, and max work the same way.

Use: one summary over everything

This aggregate has group_by: [], so it folds all the customers into a single row. Predict the two numbers, then run.

✋ Predict first

The data has three customers with lifetime_value 80000, 1200, and 4000. With group_by: [], how many output rows are there, and what are `customers` and `total_value`?

Reveal the expected result

customers,total_value
3,85200

pipeline.yaml // editable

reset ⌖ explain ✓ dry-run ▶ run

pipeline: name: customer_summary nodes: - type: source name: customers config: name: customers type: csv path: ./data/customers.csv options: { has_header: true } schema: - { name: status, type: string } - { name: lifetime_value, type: string } # An aggregate node collapses many rows into a summary. # group_by: [] means "one summary over ALL rows" (no key). - type: aggregate name: summary input: customers config: group_by: [] cxl: | emit customers = count(*) emit total_value = sum(lifetime_value.to_int()) - type: output name: results input: summary config: name: results type: csv path: ./output/summary.csv

One row out, no matter how many rows came in: that’s what group_by: [] means. customers is count(*) = 3; total_value is the sum, 80000 + 1200 + 4000 = 85200.

Modify: one row per status

Now group instead of folding everything together. In the runner above, change the aggregate to group by status and carry that column through:

group_by: [status]                 # one group per distinct status
cxl: |
  emit status = status             # the grouping key, kept in the output
  emit customers = count(*)
  emit total_value = sum(lifetime_value.to_int())

✋ Predict first

The three customers are (active, 80000), (inactive, 1200), (active, 4000). With group_by: [status], how many rows come out now, and what are the counts and totals?

Decide on your answer before you open the reveal — committing a guess is what makes it stick.

Reveal the answer

Two rows, one per distinct status:

status,customers,total_value
active,2,84000
inactive,1,1200

active groups the 80000 and 4000 customers (count 2, total 84000); inactive has just the 1200 customer. Note you must emit status = status to keep the grouping key in the output. An aggregate only emits what you ask it to.

Create: a richer per-group summary

Now author one from scratch. The question: per status, how many customers, and what is their average lifetime value? Start from the per-status aggregate above and swap the total for an average (avg is built in, like sum).

- type: aggregate
  name: by_status
  input: customers
  config:
    group_by: [status]
    cxl: |
      emit status = status                            # keep the grouping key
      emit customers = count(*)
      emit avg_value = avg(lifetime_value.to_int())   # avg, not sum

Over the same three customers, (active, 80000), (inactive, 1200), (active, 4000), that gives:

status,customers,avg_value
active,2,42000
inactive,1,1200

active averages (80000 + 4000) / 2 = 42000; inactive has just the 1200. Check your own with clinker run your_pipeline.yaml --dry-run (config + types) and --explain (the plan) before you run it for the numbers.

Grouping by time: the tumbling window

Sometimes the group you want is a slice of time: clicks per hour, sales per day. When your rows carry a timestamp, add a time_window to the aggregate. A tumbling window chops time into back-to-back, non-overlapping buckets (every hour, on the hour) and makes each bucket a group.

Here is Clinker’s real tumbling_clicks example, per-user clicks counted per hour:

- type: source
  name: clicks
  config:
    name: clicks
    type: csv
    path: ./data/clicks.csv
    options: { has_header: true }
    watermark:
      column: event_ts          # which column carries event time
    schema:
      - { name: user_id, type: string }
      - { name: event_ts, type: date_time }
      - { name: kind, type: string }

- type: aggregate
  name: hourly_clicks
  input: clicks
  config:
    group_by: [user_id]          # group by user AND by the time bucket below
    time_window:
      tumbling: { size: 1h }     # one bucket per hour
    cxl: |
      emit user_id = user_id
      emit n = count(*)

A user who clicks five times between 09:00 and 10:00 and twice between 10:00 and 11:00 produces two output rows: n = 5 for the 09:00 bucket and n = 2 for the 10:00 bucket:

user_id,n
u1,5
u1,2

🌱 New here? — watermark

A time window needs to know which column is the event time, so the source declares a watermark: { column: event_ts }. Without a watermark on the source, Clinker has no clock to bucket by, and a time_window won’t work. The column must be a date_time (or date).

Other window shapes (same idea, different slicing, and you’ll reach for these later): hopping windows overlap (e.g. a 1-hour window that advances every 15 minutes, so each row lands in several windows), and session windows group bursts of activity separated by a gap of idleness. Tumbling is the one to start with.

Retrieval checkpoint

✓ Checkpoint

// quick check

An aggregate has group_by: [status] and its CXL is `emit customers = count(*)`. The output is missing the status column. Why?

• An aggregate only emits what you ask it to; you must also `emit status = status` to keep the grouping key
• group_by automatically hides the key for privacy
• count(*) overwrites the status column

// quick check

From lesson 04: if three tenants each want clicks bucketed by a different window size (1h, 30m, 15m) but the pipeline is otherwise identical, what's the cleanest way to vary just the size?

• Declare the window size as a pipeline var and override it per tenant with a channel
• Copy the pipeline three times, once per tenant
• Put all three sizes in one aggregate and pick at runtime

Common mistakes

• Forgetting to emit the grouping key: group_by: [status] groups by status, but the output won’t contain status unless you emit status = status.
• A time window with no watermark: the source must declare watermark: { column: ... } over a date_time/date column, or the window has no clock.
• Reaching for a function that doesn’t exist: median, mode, and a bare any() group aggregate are not built in. Stick to count, sum, avg, min, max, collect, weighted_avg.

You should be able to:

• Use an aggregate with group_by: [] for one overall summary
• Use group_by: [key] for one row per group, remembering to emit the key
• Name three aggregate functions (count, sum, avg, min, max)
• Add a tumbling time_window, and say why the source needs a watermark

Verify in the checkout:

clinker run tumbling_clicks.yaml --dry-run
clinker run tumbling_clicks.yaml --explain

Go deeper (optional)

An aggregate is the first node that holds rows back. It can’t emit a group’s summary until the group is complete (for a time window, until the clock passes the bucket’s end). Next, you’ll combine separate streams together with joins and reshaping.