Windows
.qsp.window count aggregate stream into windows with equal numbers of records global aggregate stream using a custom trigger function sliding aggregate stream in potentially overlapping windows based on event time timer aggregate stream by processing time tumbling aggregate stream into non-overlapping windows based on event time
.qsp.window.count
Split the stream into equally sized windows
.qsp.window.count[size]
.qsp.window.count[size;frequency]
Parameters:
name | type | description | default |
---|---|---|---|
size | long | The exact number of records to include in each window. | Required |
frequency | long | The number of records between the starts of consecutive windows. If this is less than size , the windows will overlap. |
The window size |
For all common arguments, refer to configuring operators
This operator buffers incoming records until the number of buffered records reaches
or exceeds size
, at which point the buffer is split into windows of length size
.
Any left over records will remain in the buffer. Sliding windows can be emitted by using
the frequency
parameter. Regardless of how many overlapping windows a record will appear
in, only one copy of each record is kept in memory.
Any partial windows will be emitted on teardown.
.qsp.run
.qsp.read.fromCallback[`publish]
.qsp.window.count[10]
.qsp.write.toConsole[]
// This will emit two windows of ten records each, with five remaining in the buffer
publish ([] x: til 25)
// This will bring the number of buffered records up to ten, triggering another window
publish ([] x: til 5)
.qsp.window.global
Aggregate the stream using a custom trigger
.qsp.window.global[trigger]
.qsp.window.global[trigger; .qsp.use (!) . flip (
(`mixedSchemas; mixedSchemas);
(`state ; state)]
Parameters:
name | type | description | default |
---|---|---|---|
trigger | function | A function that splits the stream (see below). | Required |
options:
name | type | description | default |
---|---|---|---|
mixedSchemas | boolean | mixedSchemas must be true if batches are tables with different schemas. It may also significantly increase performance when batches are tuples or lists of different types. Note the caveat below about how this changes the buffer parameter. | 0b |
state | any | The initial state. | :: |
For all common arguments, refer to configuring operators
This operator splits a stream into windows by buffering incoming batches, and splitting
the buffer on the indices returned by the trigger
function.
The trigger
function is passed the following parameters:
- the operator's id
- the buffered records
- an offset of where the current batch starts
- the current batch's metadata
- the current batch's data
As batches are ingested, the trigger function will be applied to each batch, and data will be buffered. However, the buffering behavior will depend on the output of the trigger function:
- If the trigger function returns an empty list or generic null, the incoming batch will be buffered and nothing will be emitted.
- If the trigger function returns numbers, the buffer will be split on those indices, with each index being the start of a new window.
Last data batch
The last list will remain in the buffer. This last list can be emitted by returning
the count of the buffer as the last index. To map indices in the current batch to
indices in the buffer, add the offset
parameter to the indices.
The buffered records cannot be modified from the trigger function.
Batches with mixed schemas are only supported when using the mixedSchemas
option.
Caveat when using mixedSchemas
When this is set, the buffer passed to the trigger will be a list of batches, rather than a single table. The indices returned by the trigger function will still work as though the buffer were a single list.
On teardown, any records left in the buffer will be emitted as a single batch.
This pipeline emits a window whenever the high water mark is exceeded.
.qsp.run
.qsp.read.fromCallback[`publish]
.qsp.window.global[{[id; buffer; offset; md; data]
state: .qsp.get[id; md];
// Split whenever a new high water mark is seen
runningMax: or\[state , data `val];
// Find where the running maximum increases
indices: offset + 1 + where 0 <> 1 _ deltas runningMax;
.qsp.set[id; md] last runningMax;
: $[state < max data `val;
indices;
()]}; .qsp.use ``state!0 15]
.qsp.write.toConsole[]
The pipeline can then be passed data
// This will be buffered, as the initial high water mark is not exceeded
publish ([] val: 0 1 5);
// The 16 exceeds the high water mark, causing everything before it to be emitted
publish ([] val: 14 15 16);
// As each of these are the highest value seen thus far, two windows are emitted
publish ([] val: 17 18);
// With the high water mark now at 18, these will all be buffered
publish ([] val: 16 17 16);
This pipeline splits the stream on a specific value.
.qsp.run
.qsp.read.fromCallback[`publish]
.qsp.window.global[{[id; buffer; offset; md; data]
offset + where data[`status] = `new
}]
.qsp.write.toConsole[]
The pipeline can then be passed data.
Each window, after the first, will start with the status new
neg[h](`publish; ([] val: 88 89 21 98 0n 24; status: `running`running`running`running`new`running))
neg[h](`publish; ([] val: 34 28 0n 90 0n 47; status: `running`running`new`running`new`running))
.qsp.window.sliding
Aggregate the stream into potentially overlapping windows based on event time
.qsp.window.sliding[period; duration; timeColumn]
.qsp.window.sliding[period; duration; timeColumn; .qsp.use (!) . flip (
(`lateness ; lateness);
(`timeAssigner ; timeAssigner);
(`passthrough ; passthrough);
(`sort ; sort);
(`countTrigger ; countTrigger);
(`skipEmptyWindows; skipEmptyWindows))]
Parameters:
name | type | description | default |
---|---|---|---|
period | timespan, time, second or minute | How frequently windows should fire. | Required |
duration | timespan, time, second or minute | The length of each window. | Required |
timeColumn | symbol | The column containing the timestamps to window on. | Required, unless timeAssigner is specified |
options:
name | type | description | default |
---|---|---|---|
lateness | timespan, time, second or minute | Time delay before emitting a window to allow late events to arrive. | 0D |
timeAssigner | function | A time-assigner function. | None |
passthrough | boolean | Send late events through the pipeline with the next batch rather than dropping them. | 0b |
sort | boolean | Sort the window in ascending time order. | 0b |
countTrigger | long | The number of buffered records at which the buffer will be flushed automatically. | 0W |
skipEmptyWindows | boolean | Only emit non-empty windows. This can increase performance on sparse historical data. | 0b |
For all common arguments, refer to configuring operators
This operator will split the stream into windows based on the records' timestamps. Windows are based solely on the event time, not the processing time.
Period is the time between starting a new window, and duration is how long a window runs for.
- If the period and duration are the same, the result is tumbling windows.
- If the duration is longer than the period, the result will be overlapping windows.
- If the period is longer than the duration, the result is hopping windows. Hopping windows are currently not supported.
Note that to avoid running out of memory when too much late data is buffered, or
when the current window has too many records, the countTrigger
option can be used to
emit buffered records for the current window and flush late data when the number of buffered
events reaches a given threshold. Buffer counts are calculated per-key, and only the records
for the key exceeding the limit will be emitted.
A window is triggered when a timestamp is encountered past the end of that window. After that, subsequent events with a timestamp within that window will be discarded.
Any partial windows will be emitted on teardown.
This creates a pipeline that fires a window every 5 seconds, containing the previous 10 seconds of data
.qsp.run
.qsp.read.fromCallback[`publish]
.qsp.window.sliding[00:00:05; 00:00:10; `time]
.qsp.write.toConsole[]
This sends data to the pipeline
// Five minutes of historical data from an hour ago
// This will emit 60 windows when the timer fires
publish ([] time: (.z.p - 01:00) + 00:00:01 * til 300; data: 300?1f)
// Data from one minute ago
// This will be emitted as a single window when the timer fires
publish ([] time: .z.p - 00:01; data: 10?1f)
// Current window stays open till a timestamp is seen past the end of the window
// The timer will emit the entire in-progress window if there are any changes,
// potentially emitting records multiple times
time: .z.p
// This emits a window with the data points 0 1 2
publish ([] time: time; data: 0 1 2)
// Wait five seconds, then this emits a window with data points 0 1 2 3 4 5
publish ([] time: time; data: 3 4 5)
// Wait five seconds, then this emits a window with data points 0 1 2 3 4 5 6 7 8
publish ([] time: time; data: 6 7 8)
// Wait five seconds, then this emits a new window containing 9 10 11
publish ([] time: time + 00:01; data: 9 10 11)
.qsp.window.timer
Aggregate the stream by processing time
.qsp.window.timer[period]
.qsp.window.timer[period; .qsp.use (!) . flip (
(`countTrigger ; countTrigger);
(`skipEmptyWindows; skipEmptyWindows))];
Parameters:
name | type | description | default |
---|---|---|---|
period | timespan, time, second or minute | How frequently windows should fire. | Required |
options:
name | type | description | default |
---|---|---|---|
countTrigger | long | The number of buffered records at which the buffer will be flushed automatically. | 0W |
skipEmptyWindows | boolean | Only emit non-empty windows. This can increase performance when windowing sparse streams. | 0b |
For all common arguments, refer to configuring operators
Timer windows aggregate the stream by processing time, with much less overhead than event-time based windowing operators. Any data in the buffer will be emitted each period. As records are not timestamped, there is no notion of late data. Because these windows ignore event time, this will work on streams that do not have event times.
If the timer fires and the buffer is empty, an empty batch with the schema of the previous window will be emitted. However, if no batches have been received yet, then nothing will be emitted, as the schema is unknown.
The start time for each window is in the emitted batches' metadata. Due to variance in when the timer fires, window durations may not be exactly equal, as is the case for sliding and tumbling windows.
Note that to avoid running out of memory when too much data is buffered, the
countTrigger
option can be used to emit buffered records for the current window and
flush late data when the number of buffered events reaches a given threshold. Buffer
counts are calculated per-key, and only the records for the key exceeding the limit
will be emitted.
On teardown, any buffered records will be emitted.
Emit a window every 5 seconds, or when the buffered data exceeds 10,000 records:
.qsp.run
.qsp.read.fromCallback[`publish]
.qsp.window.timer[00:00:05; .qsp.use enlist[`countTrigger]!enlist 10000]
.qsp.map[count]
.qsp.write.toConsole[]
// As the timestamps are not read, these records will all be emitted together,
// within five seconds of being received.
publish ([] time: .z.p + 00:00:00 00:00:30 00:00:02; data: 3?1f);
// As the number of records exceeds the countTrigger,
// this will cause the buffer to be emitted immediately.
// As the windowing is based on processing time, timestamps are not required.
publish ([] data: 10001?1f);
.qsp.window.tumbling
Aggregate stream into non-overlapping windows based on event time
.qsp.window.tumbling[period; timeColumn]
.qsp.window.tumbling[period; timeColumn; .qsp.use (!) . flip (
(`lateness ; lateness);
(`timeAssigner ; timeAssigner);
(`passthrough ; passthrough);
(`sort ; sort);
(`countTrigger ; countTrigger);
(`skipEmptyWindows; skipEmptyWindows))]
Parameters:
name | type | description | default |
---|---|---|---|
period | timespan, time, second or minute | The length of each window. | Required |
timeColumn | symbol | The column containing the timestamps to window on. | Required, unless timeAssigner is specified |
options:
name | type | description | default |
---|---|---|---|
lateness | timespan, time, second or minute | Time delay before emitting a window to allow late events to arrive. | 0D |
timeAssigner | function | A time-assigner function. | None |
passthrough | boolean | Send late events through the pipeline with the next batch rather than dropping them. | 0b |
sort | boolean | Sort the window in ascending time order. | 0b |
countTrigger | long | The number of buffered records at which the buffer will be flushed automatically. | 0W |
skipEmptyWindows | boolean | Only emit non-empty windows. This can increase performance on sparse historical data. | 0b |
For all common arguments, refer to configuring operators
This operator will split the stream into windows based on the records' timestamps. Windows are based solely on the event time, not the processing time.
A window is triggered when a timestamp is encountered past the end of that window. After that, subsequent events with a timestamp within that window will be discarded.
Note that to avoid running out of memory when too much late data is buffered, or
when the current window has too many records, the countTrigger
option can be used to
emit buffered records for the current window and flush late data when the number of buffered
events reaches a given threshold. Buffer counts are calculated per-key, and only the records
for the key exceeding the limit will be emitted.
On teardown, any buffered records will be emitted.
Create 2-second tumbling windows:
// The `timeAssigner` pulls the time from the event-time column `time`
.qsp.run
.qsp.read.fromCallback[`publish]
.qsp.window.tumbling[00:00:10; `time]
.qsp.map[{ select avg val from x }]
.qsp.write.toConsole[]
publish ([] time: .z.p + 00:00:10 + til 100; val: 100?1f)
Time-assigner functions
If preprocessing is required to extract the times from each record or from a batch's metadata, a time assigners function can be used. The time assigner takes an incoming batch as an argument, and returns a list of timestamps or timespans with a value for every record in that batch. Alternatively, it can return a timestamp or timespan atom, to be used for all records in the batch.
This example parses the time from a string, and applies a timezone offset, without changing the value of the underlying column.
.qsp.window.tumbling[period;
.qsp.use ``timeAssigner!(::; {-05:00:00 + "P"$x`start})]
To reference metadata from the time assigner, the params
option can be used. The symbols
data
, metadata
, and operator
can be used to pass in the corresponding value for a
message's data, metadata, and the underlying operator.
.qsp.window.tumbling[period;
.qsp.use `timeAssigner`params!({[x;y;z] y `start}; `data`metadata`operator)]
Performance
Throughput is higher if batches are tables, rather than a list of tuples.
Including late data
To include late data, the acceptable lateness can be set with the lateness
option.
A new timestamp will then trigger a window only when that timestamp exceeds the sum of the
start time, the window duration, and the lateness.
This only applies to tumbling and sliding windows.
.qsp.run
.qsp.read.fromCallback[`publish]
.qsp.window.tumbling[00:00:10; `time;
.qsp.use enlist[`lateness]!enlist 00:00:02]
.qsp.write.toConsole[]
// The 1 is read one second into the second window, so it gets included in the first window.
// Once the 12 is read though, it closes out the earlier window.
// As such, the 9 will be discarded.
publish each enlist each ([] time: 0p + 00:00:01 * 0 4 3 8 10 1 12 9 20 18 25)
// If the same pattern comes in as a single batch however, all records are included
publish ([] time: 1p + 00:00:01 * 0 4 3 8 11 1 12 9 20 18 25)
Late data passthrough
By default, windows are emitted when the most recent timestamp is greater or equal to the
current window's end time + allowed lateness. Thus, with 10 seconds of lateness allowed,
a window ending at 14:00:00
will be emitted once a record is seen timestamped at or after
14:00:10
.
With passthrough enabled, windows are emitted as soon as they end, and no data will be discarded due to lateness. Any late data will be buffered until the next window is triggered, then emitted in a single batch, even when it spans more than one duration.
This only applies to tumbling and sliding windows.
time: `timestamp$`date$.z.p;
.qsp.run
.qsp.read.fromCallback[`publish]
.qsp.window.tumbling[00:00:10; `timestamp; .qsp.use enlist[`passthrough]!enlist 1b]
.qsp.write.toConsole[]
// The first window starts at time + 60 seconds
publish ([] timestamp: time + 00:00:01 * 60 68 70; val: 1 2 3)
// Late data is buffered
publish ([] timestamp: time + 00:00:01 * 0 15 30; val: 5 4 6)
// When the next window is emitted, the late data is emitted first, in a single batch
publish ([] timestamp: time + 00:00:01 * 83 84; val: 7 8)
To sort batches
To sort the emitted records ascending by timestamp, use the sort
option.
This does not apply to .qsp.window.timer
, as it has no time-assigner.
This pipeline sorts records by timestamp to calculate the deltas. State is used so the delta can be calculated across window boundaries.
.qsp.run
.qsp.read.fromCallback[`publish]
.qsp.window.tumbling[00:00:15; `timestamp; .qsp.use ``sort!11b]
.qsp.map[{[op; md; data]
previous: .qsp.get[op; md];
.qsp.set[op; md; last data `val];
1 _ deltas previous , data `val
};
.qsp.use ``state!(::; 0)]
.qsp.write.toConsole[]
shuffled: {neg[count x]?x} til 35;
publish ([] timestamp: .z.p + 00:00:01 * shuffled; val: shuffled)
Window start times
Window start times are included in the metadata for windowed batches as window
.
This only applies to tumbling and sliding windows.
This pipeline displays the average value for each window, along with when that window started:
.qsp.run
.qsp.read.fromCallback[`publish]
.qsp.window.sliding[00:00:02; 00:00:05; `timestamp]
.qsp.map[
{[md; data] select avg val, windowStart: md`window from data};
.qsp.use ``params!(::; `metadata`data)]
.qsp.write.toConsole[]
publish ([] timestamp: .z.p + 00:00:00.1 * til 200; val: 200?1f)
Temporal timeouts for windows with idle streams
For tumbling and sliding windows, buffered records are normally emitted when a new timestamp is seen past the end of the current window. In the event that a stream goes idle the window thus far will be emitted, triggered by a timer with a frequency matching the window's period.
When the timer fires, if there are buffered records and if the time since the last
window was emitted exceeds the period, all data buffered for the current window will be
emitted. As such, multiple batches can be emitted for the same window. Batches emitted by
the timer will have isPartial
set to true in their metadata.
Triggering windows on buffer size
A window will automatically be triggered when the number of buffered records
exceeds a given threshold. This also triggers a cleanup of any late data. This value can
be set using the countTrigger
option.
Note that this can lead to multiple windows being emitted with the same start time.
Batches emitted by the count trigger will have isPartial
set to true in their metadata.
.qsp.run
.qsp.read.fromCallback[`publish]
.qsp.window.tumbling[00:00:05; `timestamp;
.qsp.use (enlist `countTrigger)!enlist 10000]
.qsp.map[{count x}]
.qsp.write.toConsole[]
// As this batch exceeds the count trigger threshold, all buffered records will immediately
// be emitted, and flushed from the buffer.
publish ([] timestamp: .z.p + 00:00:00.00001 * til 35000; val: 35000?1f)
// This adds 8000 late records to the buffer
publish ([] timestamp: (.z.p - 1D) + 8000#00:00:01; val: 8000?1f)
// Adding another 2500 records triggers a cleanup of late data,
// after which the buffer size is back under the threshold, and no records will be emitted.
publish ([] timestamp: .z.p + 2500#00:00:01; val: 2500?1f)
Reducing partial windows
When a countTrigger or idle-stream timeout causes a partial window to be emitted, those windows can be aggregated using the reduce operator.