Windows & Time Operations

Module Duration: 4-5 hours Focus: Windowing in Apache Flink Prerequisites: DataStream API, Event Time concepts

Introduction

Windows are essential for bounded computations over unbounded streams. They divide the infinite stream into finite chunks based on time or count, enabling aggregations, joins, and pattern detection.

Window Types

Tumbling Windows

Non-overlapping, fixed-size windows.

// Event-time tumbling windows
DataStream<Event> input = /* ... */;

DataStream<Result> results = input
    .keyBy(e -> e.key)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .reduce(new MyReduceFunction());

// Processing-time tumbling windows
DataStream<Result> processingResults = input
    .keyBy(e -> e.key)
    .window(TumblingProcessingTimeWindows.of(Time.minutes(5)))
    .reduce(new MyReduceFunction());

// Offset windows (e.g., start at :02 instead of :00)
DataStream<Result> offsetResults = input
    .keyBy(e -> e.key)
    .window(TumblingEventTimeWindows.of(Time.minutes(5), Time.minutes(2)))
    .reduce(new MyReduceFunction());

Sliding Windows

Overlapping windows with fixed size and slide interval.

// 10-minute windows sliding every 5 minutes
DataStream<Result> slidingResults = input
    .keyBy(e -> e.key)
    .window(SlidingEventTimeWindows.of(
        Time.minutes(10),  // window size
        Time.minutes(5)    // slide interval
    ))
    .aggregate(new MyAggregateFunction());

// Processing time variant
DataStream<Result> processingSlidingResults = input
    .keyBy(e -> e.key)
    .window(SlidingProcessingTimeWindows.of(Time.minutes(10), Time.minutes(5)))
    .aggregate(new MyAggregateFunction());

Session Windows

Dynamic windows based on activity gaps.

// Session window with 10-minute gap
DataStream<Result> sessionResults = input
    .keyBy(e -> e.userId)
    .window(EventTimeSessionWindows.withGap(Time.minutes(10)))
    .process(new SessionWindowFunction());

// Dynamic gap based on element
DataStream<Result> dynamicSessionResults = input
    .keyBy(e -> e.userId)
    .window(EventTimeSessionWindows.withDynamicGap(new SessionWindowTimeGapExtractor<Event>() {
        @Override
        public long extract(Event element) {
            // Users with premium accounts get longer session gaps
            return element.isPremium() ? 30 * 60 * 1000 : 10 * 60 * 1000;
        }
    }))
    .process(new SessionWindowFunction());

Global Windows

All elements go to a single window (requires custom trigger).

DataStream<Result> globalResults = input
    .keyBy(e -> e.key)
    .window(GlobalWindows.create())
    .trigger(CountTrigger.of(100))  // Fire every 100 elements
    .process(new MyWindowFunction());

Count Windows

Windows based on element count instead of time.

// Tumbling count window: every 100 elements
DataStream<Result> countResults = input
    .keyBy(e -> e.key)
    .countWindow(100)
    .reduce(new MyReduceFunction());

// Sliding count window: 100 elements, slide every 10
DataStream<Result> slidingCountResults = input
    .keyBy(e -> e.key)
    .countWindow(100, 10)
    .reduce(new MyReduceFunction());

Window Functions

ReduceFunction

Incrementally combines elements.

public class SumReduceFunction implements ReduceFunction<SensorReading> {
    @Override
    public SensorReading reduce(SensorReading r1, SensorReading r2) {
        return new SensorReading(
            r1.sensorId,
            r1.temperature + r2.temperature,
            Math.max(r1.timestamp, r2.timestamp)
        );
    }
}

// Usage
DataStream<SensorReading> sums = readings
    .keyBy(r -> r.sensorId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .reduce(new SumReduceFunction());

AggregateFunction

More efficient than reduce, with separate accumulator.

public class AverageAggregate implements
        AggregateFunction<SensorReading, Tuple2<Double, Long>, Double> {

    @Override
    public Tuple2<Double, Long> createAccumulator() {
        return new Tuple2<>(0.0, 0L);
    }

    @Override
    public Tuple2<Double, Long> add(SensorReading reading, Tuple2<Double, Long> acc) {
        return new Tuple2<>(acc.f0 + reading.temperature, acc.f1 + 1);
    }

    @Override
    public Double getResult(Tuple2<Double, Long> acc) {
        return acc.f0 / acc.f1;
    }

    @Override
    public Tuple2<Double, Long> merge(Tuple2<Double, Long> acc1, Tuple2<Double, Long> acc2) {
        return new Tuple2<>(acc1.f0 + acc2.f0, acc1.f1 + acc2.f1);
    }
}

// Usage
DataStream<Double> averages = readings
    .keyBy(r -> r.sensorId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new AverageAggregate());

ProcessWindowFunction

Access to all window elements and metadata.

public class TopNWindowFunction extends
        ProcessWindowFunction<Event, String, String, TimeWindow> {

    private final int topN;

    public TopNWindowFunction(int topN) {
        this.topN = topN;
    }

    @Override
    public void process(
            String key,
            Context context,
            Iterable<Event> events,
            Collector<String> out) {

        List<Event> eventList = StreamSupport
            .stream(events.spliterator(), false)
            .sorted(Comparator.comparingDouble(e -> -e.value))
            .limit(topN)
            .collect(Collectors.toList());

        out.collect(String.format(
            "Key: %s, Window: [%d - %d], Top %d: %s",
            key,
            context.window().getStart(),
            context.window().getEnd(),
            topN,
            eventList
        ));
    }
}

// Usage
DataStream<String> topN = events
    .keyBy(e -> e.category)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .process(new TopNWindowFunction(10));

Combining Incremental and Full-Window Processing

// Efficient: Pre-aggregate with AggregateFunction, then process with window context
public class MyAggregateFunction implements
        AggregateFunction<Event, Long, Long> {
    // Incremental aggregation (efficient)
    // Implementation similar to above
}

public class MyProcessWindowFunction extends
        ProcessWindowFunction<Long, Result, String, TimeWindow> {
    // Has access to window metadata

    @Override
    public void process(
            String key,
            Context context,
            Iterable<Long> aggregates,
            Collector<Result> out) {

        Long aggregatedValue = aggregates.iterator().next();

        out.collect(new Result(
            key,
            aggregatedValue,
            context.window().getStart(),
            context.window().getEnd()
        ));
    }
}

// Combine both
DataStream<Result> results = input
    .keyBy(e -> e.key)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .aggregate(new MyAggregateFunction(), new MyProcessWindowFunction());

Triggers

Control when windows fire.

// Built-in triggers
// 1. EventTimeTrigger: Fires when watermark passes window end
// 2. ProcessingTimeTrigger: Fires when processing time passes window end
// 3. CountTrigger: Fires after N elements
// 4. PurgingTrigger: Fires and purges window state

// Custom trigger
public class EarlyAndLateTrigger extends Trigger<Event, TimeWindow> {

    @Override
    public TriggerResult onElement(
            Event element,
            long timestamp,
            TimeWindow window,
            TriggerContext ctx) throws Exception {

        // Register timers
        ctx.registerEventTimeTimer(window.maxTimestamp());

        // Early firing: every 1000 elements
        long count = ctx.getPartitionedState(
            new ValueStateDescriptor<>("count", Long.class)
        ).value();

        if (count != null && count >= 1000) {
            return TriggerResult.FIRE;
        }

        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onEventTime(
            long time,
            TimeWindow window,
            TriggerContext ctx) throws Exception {

        return time == window.maxTimestamp()
            ? TriggerResult.FIRE_AND_PURGE
            : TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onProcessingTime(
            long time,
            TimeWindow window,
            TriggerContext ctx) throws Exception {
        return TriggerResult.CONTINUE;
    }

    @Override
    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
        ctx.deleteEventTimeTimer(window.maxTimestamp());
        ctx.getPartitionedState(
            new ValueStateDescriptor<>("count", Long.class)
        ).clear();
    }
}

// Usage
DataStream<Result> withCustomTrigger = input
    .keyBy(e -> e.key)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .trigger(new EarlyAndLateTrigger())
    .process(new MyWindowFunction());

Evictors

Remove elements from window before/after function application.

// Built-in evictors
// 1. CountEvictor: Keeps max N elements
// 2. TimeEvictor: Keeps elements from last N milliseconds
// 3. DeltaEvictor: Keeps elements within delta of last element

// Custom evictor
public class OutlierEvictor implements Evictor<Event, TimeWindow> {

    @Override
    public void evictBefore(
            Iterable<TimestampedValue<Event>> elements,
            int size,
            TimeWindow window,
            EvictorContext evictorContext) {

        // Calculate mean
        double mean = StreamSupport.stream(elements.spliterator(), false)
            .mapToDouble(e -> e.getValue().value)
            .average()
            .orElse(0.0);

        // Calculate standard deviation
        double variance = StreamSupport.stream(elements.spliterator(), false)
            .mapToDouble(e -> Math.pow(e.getValue().value - mean, 2))
            .average()
            .orElse(0.0);
        double stdDev = Math.sqrt(variance);

        // Remove outliers (> 3 standard deviations)
        Iterator<TimestampedValue<Event>> iterator = elements.iterator();
        while (iterator.hasNext()) {
            TimestampedValue<Event> element = iterator.next();
            if (Math.abs(element.getValue().value - mean) > 3 * stdDev) {
                iterator.remove();
            }
        }
    }

    @Override
    public void evictAfter(
            Iterable<TimestampedValue<Event>> elements,
            int size,
            TimeWindow window,
            EvictorContext evictorContext) {
        // No post-processing eviction
    }
}

// Usage
DataStream<Result> withEvictor = input
    .keyBy(e -> e.key)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .evictor(new OutlierEvictor())
    .process(new MyWindowFunction());

Allowed Lateness

OutputTag<Event> lateDataTag = new OutputTag<Event>("late-events"){};

SingleOutputStreamOperator<Result> results = input
    .keyBy(e -> e.key)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .allowedLateness(Time.minutes(2))  // Keep window state for 2 extra minutes
    .sideOutputLateData(lateDataTag)
    .aggregate(new MyAggregateFunction());

// Process late data
DataStream<Event> lateData = results.getSideOutput(lateDataTag);
lateData.addSink(new LateDataSink());

Real-World Examples

Example 1: Real-Time Analytics Dashboard

public class RealTimeAnalyticsDashboard {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<PageView> pageViews = env
            .addSource(new PageViewSource())
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<PageView>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                    .withTimestampAssigner((view, ts) -> view.timestamp)
            );

        // 1-minute tumbling windows for page view counts
        DataStream<PageViewStats> stats = pageViews
            .keyBy(view -> view.page)
            .window(TumblingEventTimeWindows.of(Time.minutes(1)))
            .aggregate(new PageViewAggregator());

        // 5-minute sliding windows for trending pages
        DataStream<TrendingPage> trending = pageViews
            .keyBy(view -> view.page)
            .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))
            .process(new TrendingPageFunction());

        stats.addSink(new DashboardSink("page_views"));
        trending.addSink(new DashboardSink("trending"));

        env.execute("Real-Time Analytics Dashboard");
    }

    static class PageViewAggregator implements
            AggregateFunction<PageView, Tuple3<String, Long, Set<String>>, PageViewStats> {

        @Override
        public Tuple3<String, Long, Set<String>> createAccumulator() {
            return new Tuple3<>("", 0L, new HashSet<>());
        }

        @Override
        public Tuple3<String, Long, Set<String>> add(
                PageView view,
                Tuple3<String, Long, Set<String>> acc) {

            acc.f0 = view.page;
            acc.f1 += 1;
            acc.f2.add(view.userId);
            return acc;
        }

        @Override
        public PageViewStats getResult(Tuple3<String, Long, Set<String>> acc) {
            return new PageViewStats(acc.f0, acc.f1, acc.f2.size());
        }

        @Override
        public Tuple3<String, Long, Set<String>> merge(
                Tuple3<String, Long, Set<String>> acc1,
                Tuple3<String, Long, Set<String>> acc2) {

            acc1.f1 += acc2.f1;
            acc1.f2.addAll(acc2.f2);
            return acc1;
        }
    }
}

Example 2: User Session Analysis

public class UserSessionAnalysis {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<UserAction> actions = env
            .addSource(new UserActionSource())
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<UserAction>forBoundedOutOfOrderness(Duration.ofSeconds(30))
                    .withTimestampAssigner((action, ts) -> action.timestamp)
            );

        // Session windows with 30-minute inactivity gap
        DataStream<UserSession> sessions = actions
            .keyBy(action -> action.userId)
            .window(EventTimeSessionWindows.withGap(Time.minutes(30)))
            .process(new ProcessWindowFunction<UserAction, UserSession, String, TimeWindow>() {

                @Override
                public void process(
                        String userId,
                        Context context,
                        Iterable<UserAction> actions,
                        Collector<UserSession> out) {

                    List<UserAction> actionList = StreamSupport
                        .stream(actions.spliterator(), false)
                        .sorted(Comparator.comparingLong(a -> a.timestamp))
                        .collect(Collectors.toList());

                    long sessionDuration = context.window().getEnd() - context.window().getStart();
                    int pageViews = (int) actionList.stream()
                        .filter(a -> a.type.equals("page_view"))
                        .count();
                    boolean hasConversion = actionList.stream()
                        .anyMatch(a -> a.type.equals("purchase"));

                    out.collect(new UserSession(
                        userId,
                        context.window().getStart(),
                        context.window().getEnd(),
                        sessionDuration,
                        actionList.size(),
                        pageViews,
                        hasConversion
                    ));
                }
            });

        sessions.addSink(new SessionAnalyticsSink());
        env.execute("User Session Analysis");
    }
}

Example 3: Anomaly Detection with Windows

public class AnomalyDetection {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Metric> metrics = env
            .addSource(new MetricSource())
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<Metric>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                    .withTimestampAssigner((metric, ts) -> metric.timestamp)
            );

        // Calculate statistics over 5-minute windows
        DataStream<MetricStats> stats = metrics
            .keyBy(m -> m.metricName)
            .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))
            .aggregate(
                new StatisticsAggregator(),
                new StatsWindowFunction()
            );

        // Detect anomalies
        DataStream<Anomaly> anomalies = stats
            .keyBy(s -> s.metricName)
            .process(new AnomalyDetectionFunction());

        anomalies.addSink(new AlertingSink());
        env.execute("Anomaly Detection");
    }

    static class StatisticsAggregator implements
            AggregateFunction<Metric, StatAccumulator, StatAccumulator> {

        @Override
        public StatAccumulator createAccumulator() {
            return new StatAccumulator();
        }

        @Override
        public StatAccumulator add(Metric metric, StatAccumulator acc) {
            acc.count++;
            acc.sum += metric.value;
            acc.sumSquares += metric.value * metric.value;
            acc.min = Math.min(acc.min, metric.value);
            acc.max = Math.max(acc.max, metric.value);
            return acc;
        }

        @Override
        public StatAccumulator getResult(StatAccumulator acc) {
            return acc;
        }

        @Override
        public StatAccumulator merge(StatAccumulator acc1, StatAccumulator acc2) {
            acc1.count += acc2.count;
            acc1.sum += acc2.sum;
            acc1.sumSquares += acc2.sumSquares;
            acc1.min = Math.min(acc1.min, acc2.min);
            acc1.max = Math.max(acc1.max, acc2.max);
            return acc1;
        }
    }

    static class AnomalyDetectionFunction extends
            KeyedProcessFunction<String, MetricStats, Anomaly> {

        private ValueState<MetricStats> previousStats;

        @Override
        public void open(Configuration parameters) {
            previousStats = getRuntimeContext().getState(
                new ValueStateDescriptor<>("previous", MetricStats.class)
            );
        }

        @Override
        public void processElement(
                MetricStats current,
                Context ctx,
                Collector<Anomaly> out) throws Exception {

            MetricStats previous = previousStats.value();

            if (previous != null) {
                // Z-score anomaly detection
                double zScore = Math.abs(current.mean - previous.mean) / previous.stdDev;

                if (zScore > 3.0) {  // 3 standard deviations
                    out.collect(new Anomaly(
                        current.metricName,
                        current.mean,
                        previous.mean,
                        zScore,
                        current.windowEnd
                    ));
                }
            }

            previousStats.update(current);
        }
    }
}

Window Joins

Tumbling Window Join

DataStream<ClickEvent> clicks = /* ... */;
DataStream<ImpressionEvent> impressions = /* ... */;

DataStream<JoinedEvent> joined = clicks
    .join(impressions)
    .where(click -> click.adId)
    .equalTo(impression -> impression.adId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .apply(new JoinFunction<ClickEvent, ImpressionEvent, JoinedEvent>() {
        @Override
        public JoinedEvent join(ClickEvent click, ImpressionEvent impression) {
            return new JoinedEvent(click, impression);
        }
    });

Interval Join

// Join events that occur within a time interval
DataStream<JoinedEvent> intervalJoined = clicks
    .keyBy(click -> click.adId)
    .intervalJoin(impressions.keyBy(imp -> imp.adId))
    .between(Time.seconds(-10), Time.seconds(10))  // Click within ±10s of impression
    .process(new ProcessJoinFunction<ClickEvent, ImpressionEvent, JoinedEvent>() {
        @Override
        public void processElement(
                ClickEvent click,
                ImpressionEvent impression,
                Context ctx,
                Collector<JoinedEvent> out) {

            out.collect(new JoinedEvent(click, impression));
        }
    });

CoGroup

More flexible than join - access to all elements from both sides.

DataStream<Result> cogrouped = stream1
    .coGroup(stream2)
    .where(e1 -> e1.key)
    .equalTo(e2 -> e2.key)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .apply(new CoGroupFunction<Event1, Event2, Result>() {
        @Override
        public void coGroup(
                Iterable<Event1> first,
                Iterable<Event2> second,
                Collector<Result> out) {

            // Process all matching elements from both streams
            for (Event1 e1 : first) {
                for (Event2 e2 : second) {
                    out.collect(new Result(e1, e2));
                }
            }
        }
    });

Performance Optimization

1. Choose Right Window Function

// Best: AggregateFunction (incremental, memory efficient)
.aggregate(new MyAggregateFunction())

// Good: ReduceFunction (incremental)
.reduce(new MyReduceFunction())

// Use sparingly: ProcessWindowFunction (keeps all elements)
.process(new MyProcessWindowFunction())

// Optimal: Combine both
.aggregate(new MyAggregateFunction(), new MyProcessWindowFunction())

2. Window Size Tuning

// Too small: High overhead, many windows
.window(TumblingEventTimeWindows.of(Time.seconds(1)))

// Too large: High latency, memory pressure
.window(TumblingEventTimeWindows.of(Time.hours(24)))

// Balanced: Based on requirements
.window(TumblingEventTimeWindows.of(Time.minutes(5)))

3. State Management

// Use state TTL to prevent unbounded growth
StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.hours(1))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .build();

ValueStateDescriptor<MyState> descriptor =
    new ValueStateDescriptor<>("state", MyState.class);
descriptor.enableTimeToLive(ttlConfig);

Best Practices

1. Always Handle Late Data

OutputTag<Event> lateTag = new OutputTag<Event>("late"){};

SingleOutputStreamOperator<Result> results = stream
    .keyBy(e -> e.key)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .allowedLateness(Time.minutes(1))
    .sideOutputLateData(lateTag)
    .aggregate(new MyAggregateFunction());

results.getSideOutput(lateTag).addSink(new LateDataSink());

2. Use Appropriate Watermark Strategy

// Align watermarks with window requirements
WatermarkStrategy<Event> strategy = WatermarkStrategy
    .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(30))
    .withTimestampAssigner((event, ts) -> event.timestamp);

3. Monitor Window Latency

public class LatencyMonitoringWindowFunction extends
        ProcessWindowFunction<Event, Result, String, TimeWindow> {

    private Counter windowsFired;
    private Histogram processingLatency;

    @Override
    public void open(Configuration parameters) {
        windowsFired = getRuntimeContext()
            .getMetricGroup()
            .counter("windows_fired");

        processingLatency = getRuntimeContext()
            .getMetricGroup()
            .histogram("processing_latency", new DescriptiveStatisticsHistogram(1000));
    }

    @Override
    public void process(
            String key,
            Context context,
            Iterable<Event> elements,
            Collector<Result> out) {

        long windowEnd = context.window().getEnd();
        long currentTime = System.currentTimeMillis();
        long latency = currentTime - windowEnd;

        windowsFired.inc();
        processingLatency.update(latency);

        // Process elements
        // ...
    }
}

Exercises

Exercise 1: Implement Session Timeout Detection

Detect when users have been inactive for too long.

// TODO: Use session windows to detect user timeouts
// Alert when session duration exceeds threshold

Exercise 2: Top-N per Window

Find top N items in each window.

// TODO: Implement top-N products by sales per hour

Summary

In this module, you learned:

Window types: tumbling, sliding, session, global, count
Window functions: reduce, aggregate, process
Triggers for custom window firing logic
Evictors for element removal
Handling late data with allowed lateness
Real-world examples: analytics, sessions, anomaly detection
Window joins and CoGroup
Performance optimization techniques
Production best practices

Next Steps

Module 6: Table API & Flink SQL

Learn declarative stream processing with SQL

​Windows & Time Operations

​Introduction

​Window Types

​Tumbling Windows

​Sliding Windows

​Session Windows

​Global Windows

​Count Windows

​Window Functions

​ReduceFunction

​AggregateFunction

​ProcessWindowFunction

​Combining Incremental and Full-Window Processing

​Triggers

​Evictors

​Allowed Lateness

​Real-World Examples

​Example 1: Real-Time Analytics Dashboard

​Example 2: User Session Analysis

​Example 3: Anomaly Detection with Windows

​Window Joins

​Tumbling Window Join

​Interval Join

​CoGroup

​Performance Optimization

​1. Choose Right Window Function

​2. Window Size Tuning

​3. State Management

​Best Practices

​1. Always Handle Late Data

​2. Use Appropriate Watermark Strategy

​3. Monitor Window Latency

​Exercises

​Exercise 1: Implement Session Timeout Detection

​Exercise 2: Top-N per Window

​Summary

​Next Steps

Module 6: Table API & Flink SQL

​Additional Resources

Windows & Time Operations

Introduction

Window Types

Tumbling Windows

Sliding Windows

Session Windows

Global Windows

Count Windows

Window Functions

ReduceFunction

AggregateFunction

ProcessWindowFunction

Combining Incremental and Full-Window Processing

Triggers

Evictors

Allowed Lateness

Real-World Examples

Example 1: Real-Time Analytics Dashboard

Example 2: User Session Analysis

Example 3: Anomaly Detection with Windows

Window Joins

Tumbling Window Join

Interval Join

CoGroup

Performance Optimization

1. Choose Right Window Function

2. Window Size Tuning

3. State Management

Best Practices

1. Always Handle Late Data

2. Use Appropriate Watermark Strategy

3. Monitor Window Latency

Exercises

Exercise 1: Implement Session Timeout Detection

Exercise 2: Top-N per Window

Summary

Next Steps

Additional Resources