如何基于Kafka Stream实现按图片路径分组并统计像素颜色数量？

Hey there! Let's figure out how to get that hex code count per image sorted out. It sounds like you got stuck trying to do a second level of grouping after groupByKey()—totally understandable, since Kafka Streams handles grouping and aggregation in specific ways. Let's walk through two solid approaches to solve this, tailored to your use case.

First, Clarify the Input Message Structure #

From your description, I'll assume your input stream from the pixel-hex-topic has:

• Key: imagePath (string, e.g., /images/photo1.png)
• Value: A JSON object containing the hex code, e.g., {"hexCode": "#ffffff"}

We'll use JSON serdes for this example, but you can adapt it to custom POJOs if you prefer.

Approach 1: Use a Composite Grouping Key #

This method avoids nested aggregation by grouping on a combined key of imagePath + hexCode right off the bat. It's straightforward and efficient for this use case.

Step-by-Step Code #

1. Read the input stream:

   StreamsBuilder builder = new StreamsBuilder();
   KStream<String, JsonNode> pixelStream = builder.stream(
       "pixel-hex-topic",
       Consumed.with(Serdes.String(), Serdes.Json())
   );
   

2. Create a composite key and group:
   We'll map each message to a composite key (imagePath + separator + hexCode) and assign a count of 1 to each occurrence:

   KGroupedStream<String, Integer> groupedByImageAndHex = pixelStream
       .mapValues(value -> 1) // Mark each hex code occurrence as 1
       .groupBy(
           (imagePath, value) -> imagePath + "::" + value.get("hexCode").asText(),
           Grouped.with(Serdes.String(), Serdes.Integer())
       );
   

3. Count occurrences and format output:
   Aggregate the counts, then split the composite key to build your desired output structure:

   groupedByImageAndHex.count()
       .toStream()
       .map((compositeKey, totalCount) -> {
           // Split the composite key back into imagePath and hexCode
           String[] keyParts = compositeKey.split("::");
           String imagePath = keyParts[0];
           String hexCode = keyParts[1];
   
           // Build the output value JSON
           ObjectNode outputValue = JsonNodeFactory.instance.objectNode();
           outputValue.put("hexCode", hexCode);
           outputValue.put("count", totalCount);
   
           return new KeyValue<>(imagePath, outputValue);
       })
       .to(
           "image-hex-count-topic",
           Produced.with(Serdes.String(), Serdes.Json())
       );
   

This will output messages exactly like you described: {imagePath1: {hexCode: #fff, count:47}} (where the key is imagePath1 and the value is the hex/count object).

Approach 2: Aggregate Within Image Path Groups #

If you prefer to keep the initial grouping by imagePath, you can use aggregate to maintain a map of hex codes to counts for each image. This is useful if you need to work with the full set of counts per image at any point.

Step-by-Step Code #

1. Read and group by image path:

   StreamsBuilder builder = new StreamsBuilder();
   KStream<String, JsonNode> pixelStream = builder.stream(
       "pixel-hex-topic",
       Consumed.with(Serdes.String(), Serdes.Json())
   );
   
   KGroupedStream<String, JsonNode> groupedByImage = pixelStream.groupByKey();
   

2. Aggregate hex code counts:
   Use aggregate to build a map that tracks how many times each hex code appears per image:

   KTable<String, Map<String, Long>> hexCountTable = groupedByImage.aggregate(
       HashMap::new, // Initial value: empty map for each image
       (imagePath, hexNode, currentCountMap) -> {
           String hexCode = hexNode.get("hexCode").asText();
           // Update the count for this hex code
           currentCountMap.put(
               hexCode,
               currentCountMap.getOrDefault(hexCode, 0L) + 1
           );
           return currentCountMap;
       },
       // Materialize the state store with appropriate serdes
       Materialized.with(
           Serdes.String(),
           Serdes.map(Serdes.String(), Serdes.Long())
       )
   );
   

3. Flatten the map to individual messages:
   Convert the KTable back to a stream, then split the map into separate messages for each hex code count:

   hexCountTable.toStream()
       .flatMapValues((imagePath, hexCountMap) -> {
           List<JsonNode> outputMessages = new ArrayList<>();
           for (Map.Entry<String, Long> entry : hexCountMap.entrySet()) {
               ObjectNode outputValue = JsonNodeFactory.instance.objectNode();
               outputValue.put("hexCode", entry.getKey());
               outputValue.put("count", entry.getValue());
               outputMessages.add(outputValue);
           }
           return outputMessages;
       })
       .to(
           "image-hex-count-topic",
           Produced.with(Serdes.String(), Serdes.Json())
       );
   

Why Your Initial Approach Ran Into Trouble #

When you used groupByKey() first, you ended up with a stream grouped by imagePath—but Kafka Streams doesn't let you "re-group" within that grouped stream directly. Instead, you need to either:

• Pre-group on the composite key (image + hex) as in Approach 1, or
• Use aggregation to maintain state (like the count map) within the image path group as in Approach 2.

Both methods avoid the confusing KTable-to-KStream conversion you were stuck on, and get you the output you need.

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