# Apache Beam: Windowing

## What are Windows?

Windows are a way to group your data by their event times. But, why do you want to group on time?

So you can apply aggregations!

An example might be if you have a stream of analytic data coming from mobile phones and want to count the number of times someone opens your app **per hour**. In this scenario, you want to aggregate data within some time bounds.

## Bounded vs. Unbounded Collections

Typically, in other data processing frameworks, you only run an ETL pipeline on batches of data (possibly from a filesystem or database). However, when using Apache Beam, you can run your pipeline on **BOTH** streaming and batch data with the same code.

Bounded data is data that comes from a fixed source like a file. Unbounded data comes from a continuously updating source like a subscription. Windowing is a mechanism to turn your unbounded data stream into a bounded data set which can then be aggregated on.

## How to use Windowing Functions

You should use the windowing functions on a PCollection before you apply a grouping transform, like `GroupByKey` or `Combine`. These transforms will then group on both the timestamp (using the window) and the key (if one is present).

## Built-In Window Functions

There are 4 window functions Apache Beam provides:

1. Fixed Time Windows - Non-overlapping time intervals with consistent duration
    
2. Sliding Time Windows - Overlapping time intervals with a `period` and `window duration`
    
3. Session Windows - windows within a certain gap duration of the next element. These windows also apply on a per-key basis
    
4. Single Global Window - the default window which contains all data and no late data
    

## Example: Counting Words with Fixed Minute Windows

```java
    // Create some input data with timestamps
    List<String> inputData = Arrays.asList("foo", "bar", "foo", "foo");
    List<Long> timestamps =
        Arrays.asList(
            Duration.standardSeconds(15).getMillis(),
            Duration.standardSeconds(30).getMillis(),
            Duration.standardSeconds(45).getMillis(),
            Duration.standardSeconds(90).getMillis());

    // Create a PCollection from the input data with timestamps
    PCollection<String> items = pipeline.apply(Create.timestamped(inputData, timestamps));

    // Create a windowed PCollection
    PCollection<String> windowedItems =
        items.apply(Window.into(FixedWindows.of(Duration.standardMinutes(1))));

    PCollection<KV<String, Long>> windowedCounts = windowedItems.apply(Count.perElement());
```