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