Documentation
¶
Overview ¶
Package window provides window assigners for grouping unbounded streams into finite chunks based on time.
A WindowAssigner takes a Record's timestamp and determines which window(s) it belongs to. When a watermark passes a window's end time, that window closes and its accumulated records are emitted downstream.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type Session ¶
type Session struct {
// contains filtered or unexported fields
}
Session assigns records to variable-size windows that close after a gap of inactivity (the gap duration). Records that arrive within the gap of the previous record belong to the same session.
Session windows are dynamic — they expand when new records arrive within the gap, and merge when two sessions grow close enough to overlap.
Example with gap=30s:
Record at 00:00 → session window [00:00, 00:30) Record at 00:15 → session window expands to [00:00, 00:45) Record at 01:30 → new session window [01:30, 02:00)
func NewSession ¶
NewSession creates a session window assigner with the given inactivity gap. A session window stays open as long as records arrive within the gap. When no record arrives for gap duration, the session closes.
func (*Session) AssignWindows ¶
AssignWindows returns a single session window starting at the record's timestamp and ending at timestamp + gap. The WindowOperator merges overlapping sessions as new records arrive.
func (*Session) WindowSize ¶
WindowSize returns the session gap duration. Note: actual window size varies per session; this returns the minimum (the gap).
type Sliding ¶
type Sliding struct {
// contains filtered or unexported fields
}
Sliding assigns records to overlapping fixed-size windows that slide by a fixed interval. A single record can belong to multiple windows.
Example with size=5min, slide=1min, aligned to :00:
Window [00:00, 00:05) contains records from 00:00 to 00:04 Window [00:01, 00:06) contains records from 00:01 to 00:05 Window [00:02, 00:07) contains records from 00:02 to 00:06 ...
A record at 00:03 belongs to windows: [00:00-00:05], [00:01-00:06], [00:02-00:07], [00:03-00:08], [00:04-00:09].
func NewSliding ¶
NewSliding creates a sliding window assigner with the given window size and slide interval. Slide must be <= size.
func (*Sliding) AssignWindows ¶
AssignWindows returns all windows that the given timestamp falls into. For sliding windows, a record typically belongs to ceil(size/slide) windows.
func (*Sliding) WindowSize ¶
WindowSize returns the window duration.
type Tumbling ¶
type Tumbling struct {
// contains filtered or unexported fields
}
Tumbling assigns records to fixed-size, non-overlapping windows. Each record belongs to exactly one window.
Example with size=5min, aligned to :00:
[00:00, 00:05), [00:05, 00:10), [00:10, 00:15), ...
A record with timestamp 00:07 goes into window [00:05, 00:10).
func NewTumbling ¶
NewTumbling creates a tumbling window assigner with the given window size. Windows are aligned to Unix epoch (offset=0) by default.
func (*Tumbling) AssignWindows ¶
AssignWindows returns exactly one window for the given timestamp.
func (*Tumbling) WindowSize ¶
WindowSize returns the duration of each tumbling window.
type Window ¶
Window represents a time range [Start, End) that records are grouped into.
func MergeSessions ¶
MergeSessions merges overlapping or adjacent session windows. Two sessions overlap if sessionA.End > sessionB.Start (or vice versa). Returns the merged window if they overlap, or both windows if they don't.
type WindowAssigner ¶
type WindowAssigner interface {
// Name returns the window type name (e.g. "Tumbling", "Sliding", "Session").
Name() string
// AssignWindows returns one or more windows that the given timestamp falls into.
// A timestamp can belong to multiple windows in the case of sliding windows.
AssignWindows(timestamp time.Time) []Window
// WindowSize returns the duration of a single window.
WindowSize() time.Duration
// IsSession returns true if this assigner creates session windows.
// Session windows need merging when records arrive within the gap;
// other window types have fixed boundaries and don't merge.
IsSession() bool
}
WindowAssigner determines which window(s) a record belongs to based on its timestamp. Different window types implement this interface.
Source Files