juplo.de Git - demos/kafka/wordcount/blob - src/main/java/de/juplo/kafka/wordcount/popular/PopularStreamProcessor.java

   1 package de.juplo.kafka.wordcount.popular;
   2
   3 import lombok.extern.slf4j.Slf4j;
   4 import org.apache.kafka.common.serialization.Serdes;
   5 import org.apache.kafka.streams.*;
   6 import org.apache.kafka.streams.kstream.*;
   7 import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
   8 import org.apache.kafka.streams.state.QueryableStoreTypes;
   9 import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
  10 import org.apache.kafka.streams.state.WindowBytesStoreSupplier;
  11 import org.springframework.kafka.support.serializer.JsonSerde;
  12 import org.springframework.kafka.support.serializer.JsonSerializer;
  13
  14 import java.time.Duration;
  15 import java.util.Map;
  16 import java.util.Properties;
  17 import java.util.stream.Collectors;
  18
  19
  20 @Slf4j
  21 public class PopularStreamProcessor
  22 {
  23         public static final String TYPE = "POPULAR";
  24         public static final String KEY_VALUE_STORE_NAME = "popular";
  25         public static final String WINDOW_STORE_NAME = "popular-windows";
  26         public static final Duration WINDOW_SIZE = Duration.ofSeconds(30);
  27
  28
  29         public final KafkaStreams streams;
  30
  31
  32         public PopularStreamProcessor(
  33                         String inputTopic,
  34                         String outputTopic,
  35                         Properties properties,
  36                         WindowBytesStoreSupplier windowBytesStoreSupplier,
  37                         KeyValueBytesStoreSupplier keyValueBytesStoreSupplier)
  38         {
  39                 Topology topology = PopularStreamProcessor.buildTopology(
  40                                 inputTopic,
  41                                 outputTopic,
  42                                 windowBytesStoreSupplier,
  43                                 keyValueBytesStoreSupplier);
  44
  45                 streams = new KafkaStreams(topology, properties);
  46         }
  47
  48         static Topology buildTopology(
  49                         String inputTopic,
  50                         String outputTopic,
  51                         WindowBytesStoreSupplier windowBytesStoreSupplier,
  52                         KeyValueBytesStoreSupplier keyValueBytesStoreSupplier)
  53         {
  54                 StreamsBuilder builder = new StreamsBuilder();
  55
  56                 builder
  57                                 .stream(inputTopic, Consumed.with(inKeySerde(), inValueSerde()))
  58                                 .peek((user, userWord) -> log.info("{}: {} -> {}", inputTopic, user, userWord))
  59                                 .map((key, userWord) -> new KeyValue<>(Word.of(userWord.getWord()), Word.of(userWord.getWord())))
  60                                 .peek((key, value) -> log.info("mapped: {} -> {}", key, value))
  61                                 .groupByKey()
  62                                 .windowedBy(TimeWindows.ofSizeWithNoGrace(WINDOW_SIZE))
  63                                 .count(
  64                                                 Materialized
  65                                                                 .<Word, Long>as(windowBytesStoreSupplier)
  66                                                                 .withKeySerde(new JsonSerde<>(Word.class).noTypeInfo())
  67                                                                 .withValueSerde(Serdes.Long()))
  68                                 .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
  69                                 .toStream()
  70                                 .peek((windowedWord, counter) -> log.info("windowed: {} -> {}", windowedWord, counter))
  71                                 .map((windowedWord, counter) -> new KeyValue<>(
  72                                                 WindowedWord.of(
  73                                                                 Long.toString(windowedWord.window().startTime().getEpochSecond()),
  74                                                                 windowedWord.key().getWord()),
  75                                                 WordCounter.of(windowedWord.key().getWord(), counter)))
  76                                 .peek((windowedWord, wordCounter) -> log.info("results: {} -> {}", windowedWord, wordCounter))
  77                                 .toTable(
  78                                                 Materialized
  79                                                                 .<WindowedWord, WordCounter>as(keyValueBytesStoreSupplier)
  80                                                                 .withKeySerde(new JsonSerde<>(WindowedWord.class).noTypeInfo())
  81                                                                 .withValueSerde(new JsonSerde<>(WordCounter.class).noTypeInfo()))
  82                                 .toStream()
  83                                 .peek((windowedWord, wordCounter) -> log.info("output: {} -> {}", windowedWord, wordCounter))
  84                                 .to(outputTopic, Produced.with(outKeySerde(), outValueSerde()));
  85
  86                 Topology topology = builder.build();
  87                 log.info("\n\n{}", topology.describe());
  88
  89                 return topology;
  90         }
  91
  92         ReadOnlyKeyValueStore<WindowedWord, WordCounter> getStore()
  93         {
  94                 return streams.store(StoreQueryParameters.fromNameAndType(KEY_VALUE_STORE_NAME, QueryableStoreTypes.keyValueStore()));
  95         }
  96
  97         public void start()
  98         {
  99                 log.info("Starting Stream-Processor");
 100                 streams.start();
 101         }
 102
 103         public void stop()
 104         {
 105                 log.info("Stopping Stream-Processor");
 106                 streams.close();
 107         }
 108
 109
 110
 111         public static JsonSerde<User> inKeySerde()
 112         {
 113                 return new JsonSerde<>(User.class);
 114         }
 115
 116         public static JsonSerde<Word> inValueSerde()
 117         {
 118                 return new JsonSerde<>(Word.class);
 119         }
 120
 121         public static JsonSerde<WindowedWord> outKeySerde()
 122         {
 123                 return serde(true);
 124         }
 125
 126         public static JsonSerde<WordCounter> outValueSerde()
 127         {
 128                 return serde(false);
 129         }
 130
 131         public static <T> JsonSerde<T> serde(boolean isKey)
 132         {
 133                 JsonSerde<T> serde = new JsonSerde<>();
 134                 serde.configure(
 135                                 Map.of(JsonSerializer.TYPE_MAPPINGS, typeMappingsConfig()),
 136                                 isKey);
 137                 return serde;
 138         }
 139
 140         private static String typeMappingsConfig()
 141         {
 142                 return typeMappingsConfig(WindowedWord.class, WordCounter.class);
 143         }
 144
 145         public static String typeMappingsConfig(Class wordClass, Class wordCounterClass)
 146         {
 147                 return Map.of(
 148                                                 "word", wordClass,
 149                                                 "counter", wordCounterClass)
 150                                 .entrySet()
 151                                 .stream()
 152                                 .map(entry -> entry.getKey() + ":" + entry.getValue().getName())
 153                                 .collect(Collectors.joining(","));
 154         }
 155 }