如何实现基于固定计数的滑动窗口聚合（或转换）？

Great question! While Kafka Streams does heavily emphasize time-based windows in its official docs, you absolutely can implement a count-based sliding window aggregation (fixed size of 3 records, time-agnostic, average aggregation) using the framework's flexible Processor API and state stores. Here's a step-by-step breakdown:

Core Approach #

Instead of relying on Kafka Streams' built-in time window classes, we'll use a custom state store to track the most recent N records per key, and compute the average on each new incoming record as the window slides.

Step 1: Define a Persistent State Store #

We need a key-value store to maintain the last 3 records for each key. We'll use a Deque (double-ended queue) to efficiently add new records and remove the oldest ones once the window size is exceeded.

// Define the state store for our count-based window
StoreBuilder<KeyValueStore<String, Deque<Double>>> countWindowStore = Stores.keyValueStoreBuilder(
    Stores.persistentKeyValueStore("3-record-window-store"),
    Serdes.String(),
    // Custom Serde for Deque<Double> - you can implement this using JSON or a binary format
    Serdes.serdeFrom(new JsonSerializer<>(), new JsonDeserializer<>(Deque.class))
);

// Register the store with the StreamsBuilder
builder.addStateStore(countWindowStore);

Step 2: Use a Transformer to Manage the Window and Compute Averages #

We'll use the transform() method on our input stream to inject custom logic for updating the window and calculating the average. This gives us full control over how the window is maintained.

// Initialize your input stream
KStream<String, Double> inputStream = builder.stream(
    "input-topic",
    Consumed.with(Serdes.String(), Serdes.Double())
);

// Apply the custom transformer to build the count-based sliding window
KStream<String, Double> averageStream = inputStream.transform(
    () -> new Transformer<String, Double, KeyValue<String, Double>>() {
        private KeyValueStore<String, Deque<Double>> store;
        private static final int WINDOW_SIZE = 3;

        @Override
        public void init(ProcessorContext context) {
            // Retrieve the registered state store
            this.store = context.getStateStore("3-record-window-store");
        }

        @Override
        public KeyValue<String, Double> transform(String key, Double value) {
            // Get the existing window records for the key (or initialize a new queue)
            Deque<Double> windowRecords = store.get(key);
            if (windowRecords == null) {
                windowRecords = new LinkedList<>();
            }

            // Add the new record to the window, remove the oldest if we exceed size
            windowRecords.addLast(value);
            if (windowRecords.size() > WINDOW_SIZE) {
                windowRecords.removeFirst();
            }

            // Persist the updated window back to the state store
            store.put(key, windowRecords);

            // Calculate the average of the current window
            double average = windowRecords.stream()
                .mapToDouble(Double::doubleValue)
                .average()
                .orElse(0.0); // Fallback if no records exist (adjust based on your needs)

            // Emit the key and the computed average
            return KeyValue.pair(key, average);
        }

        @Override
        public void close() {
            // Cleanup resources if needed
        }
    },
    "3-record-window-store" // Specify the state store to use
);

// Send the results to your output topic
averageStream.to(
    "output-topic",
    Produced.with(Serdes.String(), Serdes.Double())
);

Key Considerations #

• Keyed Streams: This implementation assumes your input stream is keyed (each record has a logical key). If your stream is unkeyed, use selectKey() first to assign meaningful keys (e.g., a user ID, device ID) so windows are per entity.
• Serde for Collections: The example uses a JSON Serde for the Deque<Double> - make sure your Serde correctly serializes/deserializes the collection to avoid data corruption.
• Fault Tolerance: Kafka Streams automatically persists state store data to disk (via RocksDB by default) and replicates state changelogs, so your window state will survive restarts or failures.
• Window Behavior: This implementation slides the window one record at a time (every new record triggers a window update and average calculation). If you need to only emit results when the window is full (i.e., after 3 records), add a check to skip emitting until windowRecords.size() == WINDOW_SIZE.

内容的提问来源于stack exchange，提问作者Sebastian