Training Sets with Materialized Aggregations

---

Building training sets for windowed aggregation features is one of the most expensive parts of the offline ML pipeline. A naive point-in-time join has to scan every underlying event per spine row, which scales poorly as either your historical spine or your event table grows.

Chalk can re-use the same materialized aggregation tiles that back online serving to make offline training set construction dramatically cheaper. By binning historical events into fixed-width partial-aggregate buckets once, Chalk can then answer arbitrarily many windowed queries by merging a small number of pre-computed tiles per row — rather than re-scanning the raw events for every row of the training spine.

---

​

How it works

When a feature is defined with a materialization config, Chalk maintains a schedule that periodically backfills partial aggregate state into object storage — one row per (grouping_key, bucket_start) pair, storing the minimal state required to reconstitute the aggregation (e.g. running sum and count for a mean).

At offline query time, you can tell the planner to read directly from these tiles rather than re-aggregating the raw event table:

from chalk.client import ChalkClient

client = ChalkClient()

ds = client.offline_query(
    input={"user.id": big_input_list},
    input_times=big_input_time_list,
    output=[
        User.total_transaction_amount["7d"],
        User.total_transaction_amount["30d"],
        User.mean_transaction_amount["7d"],
    ],
    planner_options={"use_materialized_offline_query": True},
)

With use_materialized_offline_query=True, Chalk treats each spine row’s input_time as the query time, shifts the lower and upper bound of each requested window by their offsets, and aligns both bounds down to a bucket boundary. It then merges the partial aggregates of the buckets from the one that contains the lower bound up to, but not including, the one that contains the upper bound.

---

​

When this is especially useful

​

Adding a new window to an existing aggregation

Because the underlying tiles are window-agnostic, adding "14d" to a feature that previously declared "7d" and "30d" is almost free — Chalk merges a different subset of the already-materialized tiles. No new scan of the raw event table is required.

​

Deriving new features from existing state

Partial aggregate state often contains more information than the feature you originally exposed. For example, if you have materialized sum and count for a column, you can compute mean by merging those two tile streams — again without re-reading any raw events.

​

High-time-point-density backfills over heavy-hitter keys

Consider computing merchant.count_transactions_last_7d on a historical spine where transaction.timestamp has millions of events per merchant per day. A naive backfill has to, for each spine (merchant, timestamp) pair, scan every transaction within the preceding seven days — a pattern that scales roughly quadratically on heavy-hitter merchants, since both the number of spine rows and the number of events per window grow with activity.

Pre-materialized tiles flatten this. Each merchant has at most window / bucket_duration tiles overlapping any given 7-day window, so the work per spine row becomes constant, and the backfill scales linearly in the number of bucket points rather than in the product of spine rows and events per window.

---

​

Trade-off: bucket granularity coarsens accuracy

Tiles are aligned on fixed bucket boundaries (see bucket_duration in MaterializationWindowConfig). When an offline query requests a window that does not fall exactly on a bucket boundary, Chalk must decide how to treat the partial bucket at the leading or trailing edge of the window. With use_materialized_offline_query=True, Chalk aggregates whole buckets, from the one that contains the window’s lower bound up to, but not including, the one that contains its upper bound, which means:

• The bucket that contains the lower bound is included in full, so events from before the lower bound can be counted.
• The bucket that contains the upper bound is left out in full, so events from before the upper bound can be missed.
• The effective window for a row starts and ends on bucket boundaries, so it can differ from the literal window_interval requested at both ends.
• Two spine rows whose shifted bounds fall in the same buckets see identical aggregate values. Without offsets, that means rows whose input_time falls in the same bucket.

For most training-set use cases this is an acceptable trade-off given the order-of-magnitude improvement in backfill time, but you should pick a bucket_duration that is small relative to your window intervals when accuracy matters. If you need the offline computation to exactly mirror the rounding behavior of online serving, also set align_offline_chalk_window_with_materialization=True.

​

Worked example: which buckets a row uses

This example uses a feature with a three-hour window and one-hour buckets. One user has eight transactions on 2026-10-20 (UTC), and each amount is a power of two, so every sum identifies exactly which transactions were included.

Time (UTC)	09:10	09:40	10:20	11:05	11:50	12:10	12:20	12:40
Amount	1	2	4	8	16	32	64	128

from datetime import datetime, timezone

from chalk import DataFrame, Windowed, offline, windowed
from chalk.features import _, features


@features
class Transaction:
    id: int
    amount: int
    user_id: "User.id"
    timestamp: datetime


@features
class User:
    id: int
    transactions: DataFrame[Transaction]
    spend: Windowed[int] = windowed(
        "3h",
        materialization={"bucket_duration": "1h"},
        expression=_.transactions[
            _.amount,
            _.timestamp <= _.chalk_now,
            _.timestamp > _.chalk_window,
        ].sum(),
        default=0,
    )


EVENTS = [
    ("09:10", 1), ("09:40", 2), ("10:20", 4), ("11:05", 8),
    ("11:50", 16), ("12:10", 32), ("12:20", 64), ("12:40", 128),
]


@offline
def get_transactions(
    uid: User.id,
) -> DataFrame[Transaction.id, Transaction.user_id, Transaction.amount, Transaction.timestamp]:
    rows = []
    for i, (hhmm, amount) in enumerate(EVENTS, start=1):
        hour, minute = map(int, hhmm.split(":"))
        rows.append(
            Transaction(
                id=i,
                user_id=uid,
                amount=amount,
                timestamp=datetime(2026, 10, 20, hour, minute, tzinfo=timezone.utc),
            )
        )
    return DataFrame(rows)

Query the feature at four times, once with the default planner and once with use_materialized_offline_query:

from datetime import datetime, timezone

from chalk.client import ChalkClient

client = ChalkClient()
times = [
    datetime(2026, 10, 20, 12, 34, tzinfo=timezone.utc),
    datetime(2026, 10, 20, 12, 59, tzinfo=timezone.utc),
    datetime(2026, 10, 20, 12, 0, tzinfo=timezone.utc),
    datetime(2026, 10, 20, 12, 1, tzinfo=timezone.utc),
]

for planner_options in [None, {"use_materialized_offline_query": True}]:
    ds = client.offline_query(
        input={"user.id": [1] * len(times)},
        input_times=times,
        output=["user.spend__10800__"],
        recompute_features=True,
        planner_options=planner_options,
    )
    print(ds.get_data_as_pandas()["user.spend__10800__"].tolist())

Query time	Exact window	Exact sum	Buckets merged	Materialized sum
12:34	09:34 to 12:34	126	09:00, 10:00, 11:00	31
12:59	09:59 to 12:59	252	09:00, 10:00, 11:00	31
12:00	09:00 to 12:00	31	09:00, 10:00, 11:00	31
12:01	09:01 to 12:01	31	09:00, 10:00, 11:00	31

At 12:34, the lower bound is 09:34 and the upper bound is 12:34. Aligned down to the hour, they are 09:00 and 12:00. Chalk merges the 09:00, 10:00, and 11:00 buckets. The 09:00 bucket includes the 09:10 transaction, which happened before the lower bound, and the 12:00 bucket is left out, so the 12:10 and 12:20 transactions are not counted even though they happened before the upper bound. The sum is 1 + 2 + 4 + 8 + 16 = 31. All four query times fall in the 12:00 hour, so they use the same buckets and get the same value.

​

Offsets

A bound that is offset from the query time is shifted before it is aligned. This feature leaves out the most recent hour by comparing against _.chalk_now - timedelta(hours=1). Add it to the User class above:

from datetime import timedelta

lagged_spend: Windowed[int] = windowed(
    "3h",
    materialization={"bucket_duration": "1h"},
    expression=_.transactions[
        _.amount,
        _.timestamp < _.chalk_now - timedelta(hours=1),
        _.timestamp >= _.chalk_window,
    ].sum(),
    default=0,
)

At 12:34, the bounds are 09:34 and 11:34, which align to 09:00 and 11:00. Chalk merges the 09:00 and 10:00 buckets, so the sum is 1 + 2 + 4 = 7. The exact window from 09:34 to 11:34 sums to 14. At 12:59, the bounds are 09:59 and 11:59, which align to the same buckets, so the materialized sum is again 7.

---

---

​

Related reading

• Materialized Windowed Aggregations — how tiles are defined, scheduled, and served online.
• Offline Queries — other planner options for optimizing aggregations in offline queries.
• chalk aggregate backfill — CLI for triggering a backfill of aggregate tiles.