Apache Storm: невозможно получить кортежи от нескольких болтов ⇐ JAVA

[Anonymous]

Я создаю топологию, которая считывает образец видео, выполняет некоторые преобразования и сохраняет его как выходные данные, а также использую Apache Storm для одновременного применения различных фильтров. Представьте, прочитав видео, я отправил его кадры на 2 болта. один применяет эффект размытия по Гауссу, а другой повышает резкость каждого полученного кадра. И заточка, и размытие по Гауссу успешно отправляют свои кортежи в одно и то же место назначения. И теперь я хочу объединить результирующие кадры, созданные этими двумя болтами, но болт-агрегатор продолжает получать только одно значение кортежа за раз. как это исправить?

Topology.java:

import org.apache.storm.Config; импортировать org.apache.storm.LocalCluster; импортировать org.apache.storm.topology.TopologyBuilder; импортировать java.io.*; Топология публичного класса { public static void main(String[] args) выдает исключение { // Создаем файл журнала для stdout и stderr Файл logFile = новый файл("topology.log"); // Перенаправляем stdout и stderr в файл журнала PrintStream printStream = новый PrintStream (новый FileOutputStream (logFile)); попробуйте (printStream) { System.setOut(printStream); System.setErr(printStream); Конфигурация конфигурации = новая конфигурация(); // Создаем экземпляр Config для конфигурации кластера config.setDebug(истина); Построитель TopologyBuilder = новый TopologyBuilder(); // Создаем TopologyBuilder Носик носик = новый носик(); // Устанавливаем излив и болты в топологию BoltFrameAnalyzerboltFrameAnalyzer = новый BoltFrameAnalyzer(); BoltAnalysisSaverboltAnalysisSaver = новый BoltAnalysisSaver(); BoltImageProcessorboltImageProcessor = новый BoltImageProcessor(); BoltGaussianBlurboltGaussianBlur = новый BoltGaussianBlur(); BoltSharpenerboltSharpener = новый BoltSharpener(); BoltFrameAggregatorboltFrameAggregator = новый BoltFrameAggregator(); BoltOutputGeneratorboltOutputGenerator = новый BoltOutputGenerator(); builder.setSpout("носик", носик, 1); // Определить поток данных, соединив носик и болты builder.setBolt("bolt-frame-analyzer",boltFrameAnalyzer, 1).shuffleGrouping("носик"); builder.setBolt("bolt-anaанализ-saver", BoltAnalysisSaver, 1).shuffleGrouping("bolt-frame-analyzer"); builder.setBolt("bolt-image-processor",boltImageProcessor, 1).shuffleGrouping("носик"); builder.setBolt("bolt-gaussian-blur",boltGaussianBlur, 1).shuffleGrouping("bolt-image-processor"); builder.setBolt("болт-точилка", BoltSharpener, 1).shuffleGrouping("bolt-image-processor"); builder.setBolt("bolt-frame-aggregator",boltFrameAggregator, 1).shuffleGrouping("болт-точилка").shuffleGrouping("bolt-gaussian-blur"); builder.setBolt("генератор-выходных болтов", BoltOutputGenerator, 1).shuffleGrouping("агрегатор-рамок-болтов"); try (LocalCluster Cluster = new LocalCluster()) { // Используем try-with-resources кластер.submitTopology("Топология", config, builder.createTopology()); Thread.sleep(100000); // Отрегулируйте время сна по мере необходимости } // Автоматическое выключение кластера при выходе из блока try } // Закрываем PrintStream и файл журнала } } BoltGaussianBlur.java:

import org.apache.storm.topology.BasicOutputCollector; импортировать org.apache.storm.topology.OutputFieldsDeclarer; импортировать org.apache.storm.topology.base.BaseBasicBolt; импортировать org.apache.storm.tuple.Fields; импортировать org.apache.storm.tuple.Tuple; импортировать org.apache.storm.tuple.Values; импортировать org.opencv.core.Mat; импортировать org.opencv.core.Size; импортировать org.opencv.imgcodecs.Imgcodecs; импортировать org.opencv.imgproc.Imgproc; импортировать org.slf4j.Logger; импортировать org.slf4j.LoggerFactory; импортировать java.io.FileInputStream; импортировать java.io.IOException; импортировать java.util.Properties; общественный класс BoltGaussianBlur расширяет BaseBasicBolt { частные финальные кадры StringGaussianBlurFilePath; общественный BoltGaussianBlur() { пытаться { Свойства prop = новые свойства(); prop.load(new FileInputStream("config.ini")); FramesGaussianBlurFilePath = prop.getProperty("framesGaussianBlurFilePath"); } catch (IOException e) { LOG.error("BoltImageProcessor: произошла ошибка при чтении файла config.ini", e); выдать новое RuntimeException(e); } } частный статический окончательный Logger LOG = LoggerFactory.getLogger(BoltGaussianBlur.class); public void выполнить (ввод кортежа, сборщик BasicOutputCollector) { int gaussianBlurFrameNumber = input.getIntegerByField("frameNumber"); LOG.info("BoltGaussianBlur: Кадр №" + gaussianBlurFrameNumber + " успешно получен."); Mat полученныйFrame = (Mat) input.getValueByField("resizedFrame"); Мат gaussianBlurFrame = новый Mat(); Imgproc.GaussianBlur(receivedFrame, gaussianBlurFrame, новый размер(9, 9), 2, 2); Строка gaussianBlurFileName = FramesGaussianBlurFilePath + "/frame_" + gaussianBlurFrameNumber + "_gaussian_blur.png"; Imgcodecs.imwrite(gaussianBlurFileName, gaussianBlurFrame); LOG.info("BoltGaussianBlur: Кадр #" + gaussianBlurFrameNumber + " успешно преобразован в размытие по Гауссу."); Collector.emit(новые значения(gaussianBlurFrame, gaussianBlurFrameNumber)); } public void DeclareOutputFields (объявление OutputFieldsDeclarer) { Declare(new Fields("gaussianBlurFrame", "gaussianBlurFrameNumber")); } } BoltSharpener.java:

import org.apache.storm.topology.BasicOutputCollector; импортировать org.apache.storm.topology.OutputFieldsDeclarer; импортировать org.apache.storm.topology.base.BaseBasicBolt; импортировать org.apache.storm.tuple.Fields; импортировать org.apache.storm.tuple.Tuple; импортировать org.apache.storm.tuple.Values; импортировать org.opencv.core.Core; импортировать org.opencv.core.Mat; импортировать org.opencv.core.Size; импортировать org.opencv.imgproc.Imgproc; импортировать org.opencv.imgcodecs.Imgcodecs; импортировать org.slf4j.Logger; импортировать org.slf4j.LoggerFactory; импортировать java.io.FileInputStream; импортировать java.io.IOException; импортировать java.util.Properties; общественный класс BoltSharpener расширяет BaseBasicBolt { частная финальная строка FramesSharpenedFilePath; общественный BoltSharpener() { пытаться { Свойства prop = новые свойства(); prop.load(new FileInputStream("config.ini")); FramesSharpenedFilePath = prop.getProperty("framesSharpenedFilePath"); } catch (IOException e) { LOG.error("BoltSharpener: произошла ошибка при чтении файла config.ini", e); выдать новое RuntimeException(e); } } частный статический окончательный Logger LOG = LoggerFactory.getLogger(BoltSharpener.class); public void выполнить (ввод кортежа, сборщик BasicOutputCollector) { int SharpenedFrameNumber = input.getIntegerByField("frameNumber"); LOG.info("BoltSharpener: Кадр #" + SharpenedFrameNumber + " успешно получен."); Mat полученныйFrame = (Mat) input.getValueByField("resizedFrame"); Мат SharpenedFrame = новый Mat(); Imgproc.GaussianBlur(receivedFrame, SharpenedFrame, новый размер(0, 0), 10); Core.addWeighted(receivedFrame, 1,5, SharpenedFrame, -0,5, 0, SharpenedFrame); String SharpenedFileName = FrameSharpenedFilePath + "/frame_" + SharpenedFrameNumber + "_sharpened.png"; Imgcodecs.imwrite(sharpenedFileName, SharpenedFrame); LOG.info("BoltSharpener: Frame #" + SharpenedFrameNumber + " было успешно заточено."); Collector.emit(новые значения(sharpenedFrame, SharpenedFrameNumber)); } public void DeclareOutputFields (объявление OutputFieldsDeclarer) { Declare(new Fields("sharpenedFrame", "sharpenedFrameNumber")); } } BoltFrameAggregator:

import org.apache.storm.topology.BasicOutputCollector; импортировать org.apache.storm.topology.OutputFieldsDeclarer; импортировать org.apache.storm.topology.base.BaseBasicBolt; импортировать org.apache.storm.tuple.Fields; импортировать org.apache.storm.tuple.Tuple; импортировать org.apache.storm.tuple.Values; импортировать org.opencv.core.Core; импортировать org.opencv.core.Mat; импортировать org.opencv.imgcodecs.Imgcodecs; импортировать org.slf4j.Logger; импортировать org.slf4j.LoggerFactory; импортировать java.io.FileInputStream; импортировать java.io.IOException; импортировать java.util.Properties; общественный класс BoltFrameAggregator расширяет BaseBasicBolt { частный окончательный StringframesAggregated; общественный BoltFrameAggregator () { пытаться { Свойства prop = новые свойства(); prop.load(new FileInputStream("config.ini")); framesAggregated = prop.getProperty("framesAggregatedFilePath"); } catch (IOException e) { LOG.error("BoltAggregator: произошла ошибка при чтении файла config.ini", e); выдать новое RuntimeException(e); } } частный статический окончательный Logger LOG = LoggerFactory.getLogger(BoltFrameAggregator.class); public void выполнить (ввод кортежа, сборщик BasicOutputCollector) { Мат полученGaussianBlurFrame = (Mat) input.getValueByField("gaussianBlurFrame"); int gaussianBlurFrameNumber = input.getIntegerByField("gaussianBlurFrameNumber"); Мат полученSharpenedFrame = (Mat) input.getValueByField("sharpenedFrame"); int SharpenedFrameNumber = input.getIntegerByField("sharpenedFrameNumber"); intагрегированныйFrameNumber = input.getIntegerByField("sharpenedFrameNumber"); Мат агрегированныйFrame = новый Mat(); if (sharpenedFrameNumber == gaussianBlurFrameNumber) { Core.addWeighted(receivedSharpenedFrame, 1, gotGaussianBlurFrame, 1, 0,gregedFrame); String AggregatedFileName = FramesAggregated + "/frame_" + SharpenedFrameNumber + "_aggregated.png"; Imgcodecs.imwrite(aggregatedFileName,gregedFrame); LOG.info("BoltAggregator: Кадр #" + SharpenedFrameNumber + " успешно агрегирован."); Collector.emit(новые значения(aggregatedFrame, AggregatedFrameNumber)); } еще { LOG.warn("BoltAggregator: номера кадров для размытия по Гауссу и кадров с повышенной резкостью не совпадают."); } } public void DeclareOutputFields (объявление OutputFieldsDeclarer) { Declare(new Fields("aggregatedFrame", "aggregatedFrameNumber")); } } Я проверил все опечатки, попробовал присвоить имена выходным потокам размытия по Гауссу и повышения резкости. Я даже пытался сохранить принимающие кортежи, чтобы посмотреть, прибудут ли после первых 4 или 5 кортежей из gaussian-blur кортежи заточки, но это не сработало.

Журнал ошибок:
[Thread-42-bolt-frame-aggregator-executor[3, 3]] ИНФОРМАЦИЯ o.a.s.e.Executor — обработка полученного TUPLE: источник: Bolt-gaussian-blur:5, поток: по умолчанию, идентификатор: {} , [Mat [ 720*1280*CV_8UC1, isCont=true, isSubmat=false,nativeObj=0x1fc231bd4a0, dataAddr=0x1fc258f0f60], 0] PROC_START_TIME(выборка): null EXEC_START_TIME(выборка): null для TASK: 3 ... [Thread-42-bolt-frame-aggregator-executor[3, 3]] ОШИБКА o.a.s.u.Utils — Асинхронный цикл умер! java.lang.RuntimeException: java.lang.IllegalArgumentException: SharpenedFrame не существует в org.apache.storm.executor.Executor.accept(Executor.java:301) ~[storm-client-2.6.0.jar:2.6.0] в org.apache.storm.utils.JCQueue.consumeImpl(JCQueue.java:113) ~[storm-client-2.6.0.jar:2.6.0] в org.apache.storm.utils.JCQueue.consume(JCQueue.java:89) ~[storm-client-2.6.0.jar:2.6.0] в org.apache.storm.executor.bolt.BoltExecutor$1.call(BoltExecutor.java:154) ~[storm-client-2.6.0.jar:2.6.0] в org.apache.storm.executor.bolt.BoltExecutor$1.call(BoltExecutor.java:140) ~[storm-client-2.6.0.jar:2.6.0] в org.apache.storm.utils.Utils$1.run(Utils.java:398) ~[storm-client-2.6.0.jar:2.6.0] в java.base/java.lang.Thread.run(Thread.java:1583) [?:?] Вызвано: java.lang.IllegalArgumentException: SharpenedFrame не существует. в org.apache.storm.tuple.Fields.fieldIndex(Fields.java:98) ~[storm-client-2.6.0.jar:2.6.0] в org.apache.storm.tuple.TupleImpl.fieldIndex(TupleImpl.java:101) ~[storm-client-2.6.0.jar:2.6.0] в org.apache.storm.tuple.TupleImpl.getValueByField(TupleImpl.java:161) ~[storm-client-2.6.0.jar:2.6.0] в BoltFrameAggregator.execute(BoltFrameAggregator.java:33) ~[classes/:?] в org.apache.storm.topology.BasicBoltExecutor.execute(BasicBoltExecutor.java:48) ~[storm-client-2.6.0.jar:2.6.0] в org.apache.storm.executor.bolt.BoltExecutor.tupleActionFn(BoltExecutor.java:212) ~[storm-client-2.6.0.jar:2.6.0] в org.apache.storm.executor.Executor.accept(Executor.java:294) ~[storm-client-2.6.0.jar:2.6.0] ... еще 6