Apache Kafka is a popular distributed streaming platform that offers robust capabilities for handling real-time data feeds. Kafka Streams, an integral part of the Kafka ecosystem, provides a high-level stream processing library that allows for stateful and stateless processing of records in a stream. One common use case is performing aggregations over a stream of data, which can be time-windowed and results may need sending after final aggregation.

Understanding Kafka Streams Aggregation

Kafka Streams supports various forms of aggregations such as count, sum, min, max, and more complex aggregations using reduce or aggregate functions. These aggregations can be applied to records in a KStream or a KTable. For windowed aggregations, a TimeWindowedKTable results where each window is essentially a subset of records grouped by a specified time interval.

Types of Windows in Kafka Streams

Kafka Streams primarily supports three types of windows:

• Tumbling Windows: Fixed-size, non-overlapping windows
• Hopping Windows: Fixed-size, overlapping windows
• Sliding Windows: Windows that slide over time, where each window contains records within a specified duration

Sending Final Aggregation from Time-Windowed KTable

Scenario

Consider an e-commerce platform where we need to calculate the total sales per product in 1-hour windows. The final result per window should be sent downstream once the window is closed.

Implementation Steps

1. Creating a Stream and a Time-windowed KTable First, you define a source KStream from a topic, say orders, which contains order information including the product ID and the order amount.

   StreamsBuilder builder = new StreamsBuilder();
   KStream<String, Order> orders = builder.stream("orders", Consumed.with(Serdes.String(), orderSerde));

1. Grouping and Windowing The orders are then grouped by the product ID and windowed using a tumbling window definition.

1   KTable<Windowed<String>, Double> aggregatedSales = orders
2       .groupBy((key, value) -> value.getProductId(), Grouped.with(Serdes.String(), orderSerde))
3       .windowedBy(TimeWindows.of(Duration.ofHours(1)))
4       .aggregate(
5           () -> 0.0,
6           (aggKey, newValue, aggValue) -> aggValue + newValue.getAmount(),
7           Materialized.<String, Double, WindowStore<Bytes, byte[]>>as("sales-sum-store")
8               .withValueSerde(Serdes.Double())
9       );

1. Sending Final Results Once the window is closed, you want to send the results to another Kafka topic or to an external system. This can be achieved using toStream() method on KTable, which converts the result into a KStream.

1   aggregatedSales
2       .toStream()
3       .filter((windowedId, sum) -> windowedId.window().end() <= System.currentTimeMillis())
4       .map((windowedId, value) -> new KeyValue<>(windowedId.key(), value))
5       .to("aggregated-sales", Produced.with(Serdes.String(), Serdes.Double()));

The filter step ensures that records are forwarded only if the window is closed.

Summary and Key Points

The following table summarizes key components and considerations for sending final aggregation results of a time-windowed KTable:

Advanced Topics

• Suppress operator: Kafka Streams provides the suppress method to hold back records until the window closes, useful for reducing result update churn.
• Grace period: Configuring how long to wait after window end time before considering window fully closed, useful for late-arriving data.

Conclusion

Accurate windowed aggregation and timely dissemination of results are crucial for many real-time applications. Kafka Streams offers a flexible and powerful framework for such tasks, making it a preferred choice for streaming analytics.

By designing efficient Kafka Streams applications, developers can leverage real-time data processing to generate insights that are both timely and relevant, aiding in decision-making processes that are critical to business operations.