Chaining Features

What is a Chaining Feature

Chaining Features refers to features which require weaving together transformations and aggregations along with denormalization. For example, compute “average price of the last 10 items that a user purchased”. This combines an initial last_k operation to get the "last 10 items" then applies an average operation on top of it as a chained aggregation.

Or alternatively, "average weight of the last 10 items that a user purchased", but assuming that weight is not an included field in the purchases data, and rather stored in a normalized fashion in a different table.

Computing this would require:```

computing last_k items purchased per user, from the purchases stream
using the item_id to enrich the data with the item_weight
finally aggregating the item weight.

To express the above transformations, we need 2 Joins in Chronon to aggregate listings in the last 14 days and average price for these listings. Chronon now has the capability to use output of join as input to downstream computations and serve chaining features in real-time, enabling users to seamlessly access the features without worrying about upstream computation. In detail, Chronon enriches data in following three computation paradigms to provide efficient and accurate results.

Batch: Enrich a hive table containing search events with this feature to create a training set.
Streaming: Enrich a stream of search events in real-time with this feature for ML feature serving.
Application services: Synchronously respond to a “search request” with this feature as a response.

If you have similar features which would require multiple joins or groupbys, chaining features might be a good fit for your use case.

How do I use it?

You can pass in a parent join JoinSource as Source in GroupBys. For example,

# Chaining Feature API example
 
# Upstream Join. Regular chronon Join for last prices of listings
parent_join = Join(
  left=rating_driver,
  right_parts=[
    JoinPart(group_by=listing_price_last),
  ],
  online=True,
  realtime=True,
  output_namespace="chronon_test",
  team_override='ml_infra',
)
 
listing_price_last = GroupBy(
  name="chronon_test.join_right_part",
  sources=price_mutations,
  keys=["listing"],
  aggregations=[
    Aggregation(input_column="price", operation=Operation.LAST),
  ],
  output_namespace="chronon_test",
  online=True,
  team_override='ml_infra',
  accuracy=Accuracy.TEMPORAL,
)
 
# Downstream Groupby for avg price of listings viewed in last 14 days per user
# New API in bold
enriched_listings = Join(
  left = source_table,
  right_parts = [
    JoinPart(
      group_by = GroupBy(
        sources = JoinSource(join = parent_join, query=Query(...)),
        keys = ["user"],
        aggregations = [
          Aggregation(
            operation = Operation.LAST_K(100),
            input_column = "price_last",
            window = [Window(14, TimeUnit.DAYS)]
          )
        ],
        accuracy = Accuracy.TEMPORAL
      ))
  ],
  derivations = [
    Derivation(
      name = "avg_listing_view_price_last_14d",
      expression = "avg(enriched_views_price_last_last_100)"
    )
  ]
)

Configuration Examples

Chaining a Join onto a Join

A Join's left can itself be a JoinSource. Offline, the downstream join reads the upstream join's output table. Online, the fetcher serves it in two steps per request: it fetches the upstream join with the request keys, evaluates the JoinSource query's selects (and wheres) over the upstream keys and response, and then fetches the parts whose keys come from those columns. Parts keyed off the raw request are fetched concurrently with the upstream join, so chaining adds latency only to the parts that need it. The chained columns are left columns to the downstream join's derivations and logging; ts/ds are the request time, as in derivations.

# Thin parent on the same left as the ranking join: one part, the user's last 15 item ids.
recent = Join(left=spine, right_parts=[JoinPart(user_recent_items)], online=True)
ids = "user_recent_items_user_id_item_id_last15_500h"
 
# The ranking join, now chained: its existing parts and derivations are unchanged because the
# spine columns pass through by name; the new parts resolve each recent item's embedding by key.
ranking = Join(
    left=JoinSource(join=recent, query=Query(selects={
        **{c: c for c in SPINE_COLUMNS},
        **{f"recent_{i}": f"{ids}[{i - 1}]" for i in range(1, 16)},
    })),
    right_parts=[
        *existing_parts,
        JoinPart(item_embedding, prefix="candidate"),
        *[JoinPart(item_embedding, key_mapping={f"recent_{i}": "item_id"}, prefix=f"recent_{i}")
          for i in range(1, 16)],
    ],
    derivations=[*existing_derivations, Derivation("cosine_last_5", "...")],
    online=True,
)

Online requirements: the upstream join must be online=True and have its own uploaded metadata (it is resolved by name, like chained GroupBys), and callers send the upstream join's keys. Chained columns that resolve to null (a user with fewer views than slots) yield null features for the parts keyed on them rather than an error. A request the wheres filter out gets null for every chained column. fetchJoinSchema lists the chained columns in the key schema alongside the request keys because that is also the logging key schema.

Clarifications

The goal of chaining is to use output of a Join as input to downstream computations like GroupBy or a Join.
- Case 1: A Join output is the source of another GroupBy
- Case 2: A Join output is the source of another Join (offline and online, see above)
Chaining GroupBys that use a JoinSource follow the same strict right boundary as other temporal event aggregations: events with the same timestamp as the query are excluded.

Edit this page

SCD2

Sources