Modern, open-source event processing

Every expression is associated with an “entity”, allowing tables and expressions to be automatically joined. Entities eliminate redundant boilerplate code.

              # How many purchases have been made per page view
# (combines tables)?
count(purchase) / count(pageview)
            

Collect events as you move through time, and aggregate them with respect to other events. Ordered aggregation makes it easy to describe temporal interactions.

              # How many pageviews have occurred since
# the last purchase?
count(pageview, window=since(purchase))

# Auto-join purchases
# How much has been spent since the last review?
count(purchase, window=since(review))
            

Pipe syntax allows multiple operations to be chained together. Write your operations in the same order you think about them. It's timelines all the way down, making it easy to aggregate the results of aggregations.

              # What's the largest spend on food over 2 purchases?
purchase.amount 
| when(purchase.category == "food") 
| sum(window=sliding(2, $input)) # Inner aggregation
| max()                          # Outer aggregation
            

Pivot from events to time-series. Unlike grouped aggregates, generators produce rows even when there's no input, allowing you to react when something doesn't happen.

              # How many signups occurred every hour
# (even if there were none)?
count(signups, window=since(daily()))
| when(daily())
| mean()
            

Observe the value of aggregations at arbitrary points in time. Timelines are either “discrete” (instantaneous values or events) or “continuous” (values produced by a stateful aggregations). Continuous timelines let you combine aggregates computed from different event sources.

              # Compute the average review for each product?
let product_average = review
| with_key(review.product_id)
| mean()
# Each purchase joined with average
in product_average | lookup(purchase.product_id))
            

Shift values forward (but not backward) in time, allowing you to combine different temporal contexts without the risk of temporal leakage. Shifted values make it easy to compare a value “now” to a value from the past.

              # How many purchases have occurred in the last day?
let purchases_now = count(purchase)
let purchases_yesterday =
   purchases_now | shift_by(days(1))
in purchases_now - purchases_yesterday
            

It is functions all the way down. No global state, no dependencies to manage, and no spooky action at a distance. Quickly understand what a query is doing, and painlessly refactor to make it DRY.

              # How many big purchases happen each hour and where?
let cadence = hourly()
# Anything can be named and re-used
let hourly_big_purchases = purchase
| when(purchase.amount > 10)
# Filter anywhere 
| count(window=since(cadence))
# Aggregate anything
| when(cadence)
# No choosing between “when” & “having”

in {hourly_big_purchases}
# Records are just another type
| extend({
  # …modify them sequentially
  last_visit_region: last(pageview.region)
})