<commit_before>---
layout: user_guide_page
title: Aggregation and State
categories: [user-guide-page]
---

## Aggregation / State Management

This section discusses state management and fault tolerance used in windowing/streaming joins.

### Summary

Onyx provides the ability to perform updates to a state machine for segments
which are calculated over [windows] (doc/user-guide/windowing.md). For example,
a grouping task may accumulate incoming values for a number of keys over
windows of 5 minutes. This feature is commonly used for aggregations, such as
summing values, though it can be used to build more complex state machines.

#### State Example

```clojure

;; Task definition
{:onyx/name :sum-all-ages
 :onyx/fn :clojure.core/identity
 :onyx/type :function
 :onyx/group-by-key :name
 :onyx/flux-policy :recover
 :onyx/min-peers 2
 :onyx/batch-size 20}

;; Window definition
{:window/id :sum-all-ages-window
 :window/task :sum-all-ages
 :window/type :global
 :window/aggregation [:your-sum-ns/sum :age]
 :window/window-key :event-time
 :window/range [1 :hour]
 :window/doc "Adds the :age key in all segments in 1 hour fixed windows"}
```

As segments are processed, an internal state within the calculated window is
updated. In this case we are trying to sum the ages of the incoming segments.

Window aggregations are defined by a map containing the following keys:

| Key                                 | Optional? | Description                                                                                                |
|------------------------------------ | --------- | -----------------------------------------------------------------------------------------------------------|
| `:aggregation/init`                 | true      | Fn (window) to initialise the state.                                                                       |
| `:aggregation/fn`                   | false     | Fn (state, window, segment) to generate a serializable state machine update.                               |
| `:aggregation/apply-state-update`   | false     | Fn (state, entry) to apply state machine update entry to a state.                                          |
| `:aggregation/super-aggregation-fn` | true      | Fn (state-1, state-2, window) to combine two states in the case of two windows being merged.               |


In the `:window/aggregation` map in the `:sum-all-ages` window referenced above.

```clojure
(ns your-sum-ns)

(defn sum-init-fn [window]
  0)

(defn sum-application-fn [state [changelog-type value]]
  (case changelog-type
    :set-value value))

(defn sum-aggregation-fn [state window segment]
  ; k is :age
  (let [k (second (:window/aggregation window))]
    [:set-value (+ state (get segment k))]))

;; sum aggregation referenced in window definition.
(def sum
  {:aggregation/init sum-init-fn
   :aggregation/fn sum-aggregation-fn
   :aggregation/apply-state-update sum-application-fn})
    
```

Let's try processing some example segments using this aggregation:

```clojure
[{:name "John" :age 49}
 {:name "Madeline" :age 55}
 {:name "Geoffrey" :age 14}]
```

Results in the following events:

| Action           | Result                       |
|------------------|------------------------------|
| Initial state    | `0`                          |
| Incoming segment | `{:name "John" :age 49}`     |
| Changelog entry  | `[:set-value 49]`            |
| Applied to state | `49`                         |
| Incoming segment | `{:name "Madeline" :age 55}` |
| Changelog entry  | `[:set-value 104]`           |
| Applied to state | `104`                        |
| Incoming segment | `{:name "Geoffrey" :age 14}` |
| Changelog entry  | `[:set-value 128]`           |
| Applied to state | `128`                        |

This state can be emitted via triggers or another mechanism. By describing
changelog updates as a vector with a log command, such as `:set-value`
aggregation function can emit multiple types of state transition if necessary.

### Fault Tolerance

To allow for full recovery after peer crashes, the window state must be replicated
somewhere. As state updates occur, Onyx publishes the stream of changelog
updates to a replicated log.

After the changelog entry is written to the replicated log, the segment is
acked, ensuring that a segment is only cleared from the input source after the
update to window states it caused has been fully written to the log. When a
peer crash occurs, a new peer will be assigned to the task, and this peer will
play back all of the changelog entries, and apply them to the state, starting
with the initial state. As the changelog updates are read back in the same
order that they were written, the full state will be recovered. Partial updates
ensure that only minimal update data is written for each segment processed,
while remaining correct on peer failure.

### Exactly Once Data Processing

Exactly once data processing is supported via Onyx's filtering feature. When a
task's catalog has `:onyx/uniqueness-key` set, this key is looked up in the
segment and used as an ID key to determine whether the segment has been seen
before. If it has previously been processed, and state updates have been
persisted, then the segment is not re-processed. This key is persisted to the
state log along with the window changelog updates, so that previously seen keys
can be recovered in case of a peer failure.
            
#### Considerations

In order to reduce memory consumption, uniqueness-key values are persisted to a
local database, currently implemented with RocksDB. This database uses a bloom
filter, and a memory cache, allowing Onyx to avoid hitting disk
for most filter key checks.

In order to prevent unbounded increase in the size of the filter's disk
consumption, uniqueness-key values are bucketed based on recency, and the
oldest bucket is expired as the newest is filled.

Several configuration parameters are available for the rocksdb based local
filter. The most relevant of these for general configuration is
`:onyx.rocksdb.filter/num-ids-per-bucket`, and `:onyx.rocksdb.num-buckets`,
which are the size and the number of buckets referenced above.

| Parameter                                    | Description                                                                                                                                                                         | Default             |
|--------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------|
| `:onyx.rocksdb.filter/peer-block-cache-size` | RocksDB block cache size in bytes. Larger caches reduce the chance that the peer will need to check for the prescence of a uniqueness key on disk. Defaults to 100MB.               | 104857600           |
| `:onyx.rocksdb.filter/bloom-filter-bits`     | Number of bloom filter bits to use per uniqueness key value                                                                                                                         | 10                  |
| `:onyx.rocksdb.filter/num-ids-per-bucket`    | Number of uniqueness key values that can exist in a RocksDB filter bucket.                                                                                                          | 10000000            |
| `:onyx.rocksdb.filter/num-buckets`           | Number of rotating filter buckets to use. Buckets are rotated every `:onyx.rocksdb.filter/num-ids-per-bucket`, with the oldest bucket being discarded if num-buckets already exist. | 10                  |<commit_msg>Improve Exactly One Data Processing section<commit_after>---
layout: user_guide_page
title: Aggregation and State
categories: [user-guide-page]
---

## Aggregation / State Management

This section discusses state management and fault tolerance used in windowing/streaming joins.

### Summary

Onyx provides the ability to perform updates to a state machine for segments
which are calculated over [windows] (doc/user-guide/windowing.md). For example,
a grouping task may accumulate incoming values for a number of keys over
windows of 5 minutes. This feature is commonly used for aggregations, such as
summing values, though it can be used to build more complex state machines.

#### State Example

```clojure

;; Task definition
{:onyx/name :sum-all-ages
 :onyx/fn :clojure.core/identity
 :onyx/type :function
 :onyx/group-by-key :name
 :onyx/flux-policy :recover
 :onyx/min-peers 2
 :onyx/batch-size 20}

;; Window definition
{:window/id :sum-all-ages-window
 :window/task :sum-all-ages
 :window/type :global
 :window/aggregation [:your-sum-ns/sum :age]
 :window/window-key :event-time
 :window/range [1 :hour]
 :window/doc "Adds the :age key in all segments in 1 hour fixed windows"}
```

As segments are processed, an internal state within the calculated window is
updated. In this case we are trying to sum the ages of the incoming segments.

Window aggregations are defined by a map containing the following keys:

| Key                                 | Optional? | Description                                                                                                |
|------------------------------------ | --------- | -----------------------------------------------------------------------------------------------------------|
| `:aggregation/init`                 | true      | Fn (window) to initialise the state.                                                                       |
| `:aggregation/fn`                   | false     | Fn (state, window, segment) to generate a serializable state machine update.                               |
| `:aggregation/apply-state-update`   | false     | Fn (state, entry) to apply state machine update entry to a state.                                          |
| `:aggregation/super-aggregation-fn` | true      | Fn (state-1, state-2, window) to combine two states in the case of two windows being merged.               |


In the `:window/aggregation` map in the `:sum-all-ages` window referenced above.

```clojure
(ns your-sum-ns)

(defn sum-init-fn [window]
  0)

(defn sum-application-fn [state [changelog-type value]]
  (case changelog-type
    :set-value value))

(defn sum-aggregation-fn [state window segment]
  ; k is :age
  (let [k (second (:window/aggregation window))]
    [:set-value (+ state (get segment k))]))

;; sum aggregation referenced in window definition.
(def sum
  {:aggregation/init sum-init-fn
   :aggregation/fn sum-aggregation-fn
   :aggregation/apply-state-update sum-application-fn})
    
```

Let's try processing some example segments using this aggregation:

```clojure
[{:name "John" :age 49}
 {:name "Madeline" :age 55}
 {:name "Geoffrey" :age 14}]
```

Results in the following events:

| Action           | Result                       |
|------------------|------------------------------|
| Initial state    | `0`                          |
| Incoming segment | `{:name "John" :age 49}`     |
| Changelog entry  | `[:set-value 49]`            |
| Applied to state | `49`                         |
| Incoming segment | `{:name "Madeline" :age 55}` |
| Changelog entry  | `[:set-value 104]`           |
| Applied to state | `104`                        |
| Incoming segment | `{:name "Geoffrey" :age 14}` |
| Changelog entry  | `[:set-value 128]`           |
| Applied to state | `128`                        |

This state can be emitted via triggers or another mechanism. By describing
changelog updates as a vector with a log command, such as `:set-value`
aggregation function can emit multiple types of state transition if necessary.

### Fault Tolerance

To allow for full recovery after peer crashes, the window state must be replicated
somewhere. As state updates occur, Onyx publishes the stream of changelog
updates to a replicated log.

After the changelog entry is written to the replicated log, the segment is
acked, ensuring that a segment is only cleared from the input source after the
update to window states it caused has been fully written to the log. When a
peer crash occurs, a new peer will be assigned to the task, and this peer will
play back all of the changelog entries, and apply them to the state, starting
with the initial state. As the changelog updates are read back in the same
order that they were written, the full state will be recovered. Partial updates
ensure that only minimal update data is written for each segment processed,
while remaining correct on peer failure.

### Exactly Once Aggregation Updates

Exactly once aggregation updates are supported via Onyx's filtering feature. When a
task's catalog has `:onyx/uniqueness-key` set, this key is looked up in the
segment and used as an ID key to determine whether the segment has been seen
before. If it has previously been processed, and state updates have been
persisted, then the segment is not re-processed. This key is persisted to the
state log transactionally with the window changelog updates, so that previously seen keys
can be recovered in case of a peer failure.

**See the section "Exactly Once Side-Effects" for discussion of why
side-effects are impossible to achieve Exactly Once**.
            
#### Considerations

In order to reduce memory consumption, uniqueness-key values are persisted to a
local database, currently implemented with RocksDB. This database uses a bloom
filter, and a memory cache, allowing Onyx to avoid hitting disk
for most filter key checks.

In order to prevent unbounded increase in the size of the filter's disk
consumption, uniqueness-key values are bucketed based on recency, and the
oldest bucket is expired as the newest is filled.

Several configuration parameters are available for the rocksdb based local
filter. The most relevant of these for general configuration is
`:onyx.rocksdb.filter/num-ids-per-bucket`, and `:onyx.rocksdb.num-buckets`,
which are the size and the number of buckets referenced above.

| Parameter                                    | Description                                                                                                                                                                         | Default             |
|--------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------|
| `:onyx.rocksdb.filter/peer-block-cache-size` | RocksDB block cache size in bytes. Larger caches reduce the chance that the peer will need to check for the prescence of a uniqueness key on disk. Defaults to 100MB.               | 104857600           |
| `:onyx.rocksdb.filter/bloom-filter-bits`     | Number of bloom filter bits to use per uniqueness key value                                                                                                                         | 10                  |
| `:onyx.rocksdb.filter/num-ids-per-bucket`    | Number of uniqueness key values that can exist in a RocksDB filter bucket.                                                                                                          | 10000000            |
| `:onyx.rocksdb.filter/num-buckets`           | Number of rotating filter buckets to use. Buckets are rotated every `:onyx.rocksdb.filter/num-ids-per-bucket`, with the oldest bucket being discarded if num-buckets already exist. | 10                  |
{'fraction_non_alphanumeric': 0.12788233027169374, 'fraction_numerical': 0.007964544929025628, 'mean_word_length': 1.9206527855936972, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 57, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2227009', 'n_tokens_mistral': 3903, 'n_tokens_neox': 3628, 'n_words': 1674}