Блог

Я новичок в Apache Storm.

Я пытаюсь разработать систему потоковой обработки в реальном времени, используя Apache Kafka, Storm и ESPER CEP engine.

Для этого у меня есть один KafkaSpout, который будет отправлять потоки в Bolts (в котором есть мои запросы CEP) для фильтрации потока.

Я уже создал топологию и пытаюсь запустить ее в локальном кластере

Проблема в том, что запрос CEP, выполняемый в моих bolts, требует пакетов кортежей для выполнения оконных операций над потоками. И в моей топологии KafkaSpout отправляет Болтам для обработки только один кортеж за раз. Итак, мой запрос CEP работает не так, как ожидалось.

Я использую KafkaSpout по умолчанию в Storm. Есть ли какой-либо способ, которым я могу отправить несколько разных кортежей одновременно в Bolts? Некоторые настройки конфигурации могут это сделать или мне нужно создать свой собственный KafkaSpout для этого?

Пожалуйста, помогите!!

Моя топология:

TopologyBuilder builder = новый TopologyBuilder();

builder.setSpout(«KafkaSpout», новый KafkaSpout<>(KafkaSpoutConfig.builder(«localhost:» 9092, «weatherdata»).setProp(ConsumerConfig.GROUP_ID_CONFIG, «weather-consumer-group»).build()),4);

builder.setBolt(«A», новые FeatureSelectionBolt(), 2).globalGrouping(«KafkaSpout»);

builder.setBolt(«B», новый TrendDetectionBolt(), 2).Группировка в случайном порядке(«A»)

Я использую 2 Bolts и один spout.

Мой запрос esper, выполняемый в Bolt A, является

выберите первый (e), последний (e) из weatherEvent.win:длина (3) как e

Здесь я пытаюсь получить первое и последнее событие из окна длиной три из потока событий. Но я получаю одно и то же первое и последнее событие, потому что KafkaSpout отправляет только один кортеж за раз.

Spout не может этого сделать, но вы можете использовать любую поддержку окон Storm https://storm.apache.org/releases/2.0.0-SNAPSHOT/Windowing.html или просто напишите модуль агрегации bolt и поместите его между spout и остальной топологией.

Итак, ваша топология должна быть spout -> aggregator -> feature selection -> trend detection .

Я бы порекомендовал вам попробовать встроенную поддержку работы с окнами, но если вы предпочитаете написать свою собственную агрегацию, вашему bolt действительно просто нужно получить некоторое количество кортежей (например, 3) и выдать новый кортеж, содержащий все значения.

Агрегатор bolt должен выполнять что-то вроде

 private List<Tuple> buffered;

execute(Tuple input) {
  if (buffered.size != 2) {
    buffered.add(input)
    return
  }
  Tuple first = buffered.get(0)
  Tuple second = buffered.get(1)
  Values aggregate = new Values(first.getValues(), second.getValues(), input.getValues())
  List<Tuple> anchors = List.of(first, second, input)
  collector.emit(anchors, aggregate)
  collector.ack(first, second, input)
  buffered.clear()
}
  

Таким образом, вы получите один кортеж, содержащий содержимое 3 входных кортежей.