Windowing

/**
 * {@link Window} logically divides up or groups the elements of a {@link PCollection} 
 * into finite windows according to a {@link WindowFn}. The output of {@code Window} 
 * contains the same elements as input, but they have been logically assigned to windows.
 *
 * Windowing a {@link PCollection} divides the elements into windows based on the 
 * associated event time for each element. This is especially useful for PCollections 
 * with unbounded size, since it allows operating on a sub-group of the elements placed 
 * into a related window.
 *
 * Several predefined {@link WindowFn}s are provided:
 * - {@link FixedWindows} partitions the timestamps into fixed-width intervals.
 * - {@link SlidingWindows} places data into overlapping fixed-width intervals.
 * - {@link Sessions} groups data into sessions where each item in a window is 
 *   separated from the next by no more than a specified gap.
 */

import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.joda.time.Duration;

// 5-minute fixed windows
PCollection<String> windowed = input.apply(
    Window.<String>into(FixedWindows.of(Duration.standardMinutes(5))));

// 1-hour windows
PCollection<String> hourly = input.apply(
    Window.<String>into(FixedWindows.of(Duration.standardHours(1))));

import org.apache.beam.sdk.transforms.windowing.SlidingWindows;

// 10-minute windows, starting every 5 minutes
PCollection<String> windowed = input.apply(
    Window.<String>into(
        SlidingWindows.of(Duration.standardMinutes(10))
            .every(Duration.standardMinutes(5))));

import org.apache.beam.sdk.transforms.windowing.Sessions;

// Session windows with 10-minute gap
PCollection<String> sessions = input.apply(
    Window.<String>into(
        Sessions.withGapDuration(Duration.standardMinutes(10))));

import org.apache.beam.sdk.transforms.windowing.GlobalWindows;

// Explicitly specify global windows (usually not needed)
PCollection<String> global = input.apply(
    Window.<String>into(new GlobalWindows()));

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.joda.time.Duration;

Pipeline p = Pipeline.create(options);

// Read streaming data
PCollection<String> input = p.apply(
    PubsubIO.readStrings().fromTopic("projects/myproject/topics/traffic"));

// Apply 1-hour fixed windows
PCollection<String> windowed = input.apply(
    Window.<String>into(FixedWindows.of(Duration.standardHours(1))));

// Count events per window
PCollection<KV<String, Long>> counts = windowed
    .apply("ExtractRoute", ParDo.of(new ExtractRouteFn()))
    .apply("CountPerRoute", Count.perElement());

// Write results
counts.apply("FormatResults", ParDo.of(new FormatResultsFn()))
      .apply(BigQueryIO.write()...);

p.run();

class AddWindowInfoFn extends DoFn<String, String> {
    @ProcessElement
    public void processElement(
            @Element String element,
            BoundedWindow window,
            OutputReceiver<String> out) {
        
        String windowInfo = String.format(
            "Element '%s' in window [%s..%s)",
            element,
            window.maxTimestamp().minus(window.maxTimestamp()),
            window.maxTimestamp());
        
        out.output(windowInfo);
    }
}

import org.apache.beam.sdk.transforms.windowing.AfterWatermark;
import org.apache.beam.sdk.transforms.windowing.Window;

PCollection<String> windowed = input.apply(
    Window.<String>into(FixedWindows.of(Duration.standardMinutes(5)))
        .withAllowedLateness(Duration.standardMinutes(10))
        .discardingFiredPanes());

import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;

PCollection<KV<String, Iterable<Integer>>> grouped = pairs.apply(
    Window.<KV<String, Integer>>into(FixedWindows.of(Duration.standardMinutes(5)))
        .withTimestampCombiner(TimestampCombiner.EARLIEST)
    .apply(GroupByKey.create()));

// Count events per 5-minute window
PCollection<Long> counts = input
    .apply(Window.into(FixedWindows.of(Duration.standardMinutes(5))))
    .apply(Count.globally().withoutDefaults());

// Sum values per key per hour
PCollection<KV<String, Integer>> sums = keyedValues
    .apply(Window.into(FixedWindows.of(Duration.standardHours(1))))
    .apply(Sum.integersPerKey());

// Average per 10-minute sliding window
PCollection<Double> averages = numbers
    .apply(Window.into(
        SlidingWindows.of(Duration.standardMinutes(10))
            .every(Duration.standardMinutes(5))))
    .apply(Mean.globally());

Pipeline p = Pipeline.create(options);

// Read click events
PCollection<ClickEvent> clicks = p.apply(
    PubsubIO.read(ClickEvent.class).fromTopic("clicks"));

// Group into user sessions (30-minute inactivity gap)
PCollection<KV<String, ClickEvent>> userClicks = clicks.apply(
    ParDo.of(new ExtractUserIdFn()));

PCollection<KV<String, Iterable<ClickEvent>>> sessions = userClicks
    .apply(Window.into(Sessions.withGapDuration(Duration.standardMinutes(30))))
    .apply(GroupByKey.create());

// Calculate session metrics
PCollection<SessionMetrics> metrics = sessions.apply(
    ParDo.of(new ComputeSessionMetricsFn()));

metrics.apply(BigQueryIO.write()...);
p.run();