Блог

Я использую потоки Kafka для чтения из темы в моем кластере, и я хочу фильтровать сообщения на основе его содержимого в формате JSON, т.Е.:

Формат JSON:

 {
   "id": 1 
   "timestamp": "2019-04-21 12:53:18", 
   "priority": "Medium", 
   "name": "Sample Text",
   "metadata": [{
      "metric_1": "0", 
      "metric_2": "1", 
      "metric_3": "2"
   }]
}
  

Я хочу прочитать сообщения из темы ввода (назовем ее «input-topic»), отфильтровать их (предположим, мне нужны только сообщения с приоритетом «Low»), затем объединить их и отправить в другую тему («filtered-topic»)

У меня не так много кода, кроме создания самого потока и его конфигураций. Я думаю, что в Serdes должно быть что-то, что мне нужно настроить, но я не уверен, как. Я также пытался использовать десериализатор JSON, но не смог заставить его работать.

Прежде всего, возможно ли это вообще? Если да, то каков будет правильный курс действий?

Вы можете создать поток из темы.

     StreamsBuilder builder = new StreamsBuilder();

    // key value type here is both String for me and update based on cases
    KStream<String, String> source = builder.stream("input-topic");

    source.filter(new Predicate<String, String>() {
        @Override
        public boolean test(String s, String s2) {
            // your filter logic here and s and s2 are key/value from topic
            // In your case, s2 should be type of your json Java object
            return false;
        }
    }).groupBy(new KeyValueMapper<String, String, String>() {
        @Override
        public String apply(String key, String value) {
            // your group by logic
            return null;
        }
    }).count().toStream().to("new topic");