Stream

Stream (stream)

Track per-key changes and emit deltas, elapsed times, and optional aggregates.

Transform json

Minimal example

actions:
  - stream:
      watch: ""

JSON

{
  "actions": [
    {
      "stream": {
        "watch": ""
      }
    }
  ]
}

Contents

• Minimal example

• Fields

• General

• Behaviour

• Input

• Output

• Missing Data

• Recency

• Aggregations

• Windowing

• Resources

• Schema

Fields

Field	Type	Required	Description
watch Input	field (string)	✅	Field whose value should be tracked. Examples: data_field
description General	string		Describe this step.
condition General	lua-expression (string)		Only run this action when the Lua condition evaluates to true. Examples: 2 * count()
delta Behaviour	boolean (bool)		Emit the difference between the current and last value.
marker Output	string		Optional marker added to generated events (when not enriching).
only-changes Behaviour	boolean (bool)		Only emit records when the watched field changes.
group-by Input	field (string)		Group tracking state by this field’s value. Examples: data_field
input-time Input	field (string)		Source field providing event time for elapsed calculations. Examples: data_field
output-field Output	field (string)		Field name used for the emitted delta. Examples: data_field
elapsed-field Output	field (string)		Field name capturing elapsed milliseconds. Examples: data_field
fill-strategy Missing Data	Fill Strategy		Strategy applied when the watched value is missing.
recency Recency	Recency		Emit a boolean indicating whether the key changed within a threshold.
aggregations Aggregations	Aggregations[]		Optional incremental aggregates maintained per key.
reset-on-document-end Behaviour	boolean (bool)		Reset state when a document end marker is observed.
window-size Windowing	number (integer)		Maximum number of recent events to keep per key when computing aggregates. Examples: 42, 1.2e-10
max-keys Resources	number (integer)		Maximum number of concurrent keys tracked. Examples: 42, 1.2e-10
eviction Resources	Eviction		Eviction policy applied when max-keys is exceeded.

General

Show fields

Field	Type	Required	Description
description	string		Describe this step.
condition	lua-expression (string)		Only run this action when the Lua condition evaluates to true. Examples: 2 * count()

Behaviour

Show fields

Field	Type	Required	Description
delta	boolean (bool)		Emit the difference between the current and last value.
only-changes	boolean (bool)		Only emit records when the watched field changes.
reset-on-document-end	boolean (bool)		Reset state when a document end marker is observed.

Input

Show fields

Field	Type	Required	Description
watch	field (string)	✅	Field whose value should be tracked. Examples: data_field
group-by	field (string)		Group tracking state by this field’s value. Examples: data_field
input-time	field (string)		Source field providing event time for elapsed calculations. Examples: data_field

Output

Show fields

Field	Type	Required	Description
marker	string		Optional marker added to generated events (when not enriching).
output-field	field (string)		Field name used for the emitted delta. Examples: data_field
elapsed-field	field (string)		Field name capturing elapsed milliseconds. Examples: data_field

Missing Data

Show fields

Field	Type	Required	Description
fill-strategy	Fill Strategy		Strategy applied when the watched value is missing.

Recency

Show fields

Field	Type	Required	Description
recency	Recency		Emit a boolean indicating whether the key changed within a threshold.

Aggregations

Show fields

Field	Type	Required	Description
aggregations	Aggregations[]		Optional incremental aggregates maintained per key.

Windowing

Show fields

Field	Type	Required	Description
window-size	number (integer)		Maximum number of recent events to keep per key when computing aggregates. Examples: 42, 1.2e-10

Resources

Show fields

Field	Type	Required	Description
max-keys	number (integer)		Maximum number of concurrent keys tracked. Examples: 42, 1.2e-10
eviction	Eviction		Eviction policy applied when max-keys is exceeded.

Schema

• Fill Strategy Options
• Eviction Options
• Fill Strategy - Default Value Fields
• Recency Fields
• Aggregations Fields
• Eviction - Ttl Fields
• Aggregations - Op Options

Fill Strategy Options

Option	Name	Type	Description
forward-fill	Forward Fill	map	Carry forward the previous value when the field is missing.
default-value	Default Value	object	Substitute a default JSON value when the field is missing.

Eviction Options

Option	Name	Type	Description
drop-new	Drop New	map	Drop new keys when the limit is reached.
lru	Lru	map	Evict the least-recently updated key.
ttl	Ttl	object	Evict keys that have been idle longer than the configured duration.

Fill Strategy - Default Value Fields

Field	Type	Required	Description
value	map (object)	✅

Recency Fields

Field	Type	Required	Description
threshold-ms	duration (integer)	✅	Threshold in milliseconds for considering a key “recent”.
output-field	field (string)		Field where the recency flag is written. Examples: data_field

Aggregations Fields

Field	Type	Required	Description
field	field (string)	✅	Source field used for the aggregation. Examples: data_field
op	Op		Aggregation operation. Allowed values: avg, mean, min, max, first, last, sum, stddev, …
r-as	field (string)		Optional alias for the output field. Examples: data_field
window-size	number (integer)		Override window size for this aggregation (falls back to stream-level window). Examples: 42, 1.2e-10
percentile	number (integer)		Percentile to compute when op = percentile (0-100). Examples: 42, 1.2e-10

Eviction - Ttl Fields

Field	Type	Required	Description
max-idle	string	✅

Aggregations - Op Options

Value	Description
avg	Avg
mean	Mean
min	Min
max	Max
first	First
last	Last
sum	Sum
stddev	Stddev
range	Range
earliest	Earliest
latest	Latest
count	Count
distinct-count	Distinct Count
median	Median
percentile	Percentile
variance	Variance
variance-population	Variance Population
stddev-population	Stddev Population
sum-squares	Sum Squares
mode	Mode
list	List
values	Values