Flink WindowAssigner assigns a window for every arrived element?

user_7194392 0
Viewed 35

why TumblingProcessingTimeWindows assigns a window for every arrived element code as below? For example, a TimeWindow with starttime of 1s and endtime 5s, then all elements between the time are expected to one window, but from the code below, every element gets a new window, why?

public class TumblingProcessingTimeWindows extends WindowAssigner<Object, TimeWindow> {
    @Override
    public Collection<TimeWindow> assignWindows(Object element, long timestamp, WindowAssignerContext context) {
        final long now = context.getCurrentProcessingTime();
        long start = TimeWindow.getWindowStartWithOffset(now, offset, size);
        return Collections.singletonList(new TimeWindow(start, start + size));
    }
}

WindowOperator invoke windowAssigner.assignWindows for every element, why:

WindowOperator.java
    @Override
    public void processElement(StreamRecord<IN> element) throws Exception {
        final Collection<W> elementWindows = windowAssigner.assignWindows(
            element.getValue(), element.getTimestamp(), windowAssignerContext);
}
apache-flink
1 Answers

That's an artifact of how the implementation was done.

What ultimately matters is how a window's contents are stored in the state backend. Flink's state backends are organized around triples: (key, namespace, value). For a keyed time window, what gets stored is

  • key: the key
  • namespace: a copy of the time window (i.e., its class, start, and end)
  • value: the list of elements assigned to this window pane

The TimeWindow object is just a convenient wrapper holding together the identifying information for each window. It's not a container used to store the elements being assigned to the window.

The code involved is pretty complex, but if you want to jump into the heart of it, you might take a look at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator#processElement (and also EvictingWindowOperator#processElement, which is very similar). Those methods use keyed state to store each incoming event in the window like this:

windowState.setCurrentNamespace(stateWindow);
windowState.add(element.getValue());

where windowState is

/** The state in which the window contents is stored. Each window is a namespace */
private transient InternalAppendingState<K, W, IN, ACC, ACC> windowState;

InternalAppendingState is a variant of ListState that exposes the namespace (which Flink's public APIs don't provide access to).

user_20008230
Related