Я хочу поместить потоковый SQL в Kafka для использования Flink для CEP. Это хороший способ? Я знаю, что определение динамического шаблона не разрешено во Flink, и мне нужно применить правила, которые они могут изменить в неограниченном потоке...

Я отправляю поток данных с виртуальной машины в тестовую тему Kafka (работает на хост-ОС с IP-адресом 192.168.0.12), используя приведенный ниже код. public class WriteToKafka { public static void main(String[] args) throws Exception {...

Я имею дело с потоком мутаций базы данных, то есть с потоком журнала изменений. Я хочу преобразовать значения с помощью SQL-запроса. Мне трудно собрать вместе следующие три концепции RowTypeInfo , Row и DataStream . ПРИМЕЧАНИЕ: я заранее не...

Из объявления о выпуске Flink 1.5 мы знаем, что Flink теперь поддерживает «широковещательное состояние», и было описано, что «широковещательное состояние разблокирует реализацию функции« динамических шаблонов »для библиотеки Flink CEP». Означает ли...

Как указано в документе CEP ( https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/libs/cep.html ), что в последовательности шаблонов допускается только одно временное ограничение, я изо всех сил пытаюсь найти способ обработки...

Предположим, у меня есть поток торговых событий на фондовом рынке, например: technical1, ALXN, 1/1/2016 technical1, CELG, 1/1/2016 technical2, ALXN, 1/2/2016 technical2, CELG, 1/2/2016 . . . technicalN, ALXN, 4/1/2018 technicalN, CELG, 4/1/2018...

Я исследую Flink уже больше недели. Мы потребляем события из Kafka и хотим, чтобы события принадлежали определенному идентификатору объекта, который необходимо обработать в порядке времени события. Пока мое исследование подсказывает мне, что я...

Допустим, у меня есть inputStream, и я выполняю над ним некоторую оконную операцию. Какая метка времени для события, созданного при выполнении над ним некоторой оконной операции. .... DataStream<Integer> inputStream = // ......

Я запускаю несколько заданий из моего файла .jar. Я хочу разделить состояние между моими работами. но все входы потребляют (из кафки) в каждой работе и генерируют дублирующийся выход. Я вижу свою мигающую панель. все вакансии «отправлено» 3. Я...

Я написал часть Flink CEP, которая проверяет шаблон статуса (с ключом id ) с Relaxed Contiguity ( followedBy ). Идея состоит в том, чтобы поднять предупреждение, если определенный статус не прибыл после первого в течение заданного времени. Это...

Учитывая, У меня есть задание Flink, которое читает из ActiveMQ источника и записывает в базу данных mysql - с ключом по идентификатору. Я включил контрольные точки для этой работы каждую секунду. Я указываю контрольные точки на экземпляр...

У меня есть интересный сценарий, в котором я работаю над сопоставлением шаблонов во flink, оценивая входящие шаблоны с помощью функции keyedbroadcastprocess, когда я запускаю программу в среде IDE, я получаю исключение нулевого указателя в функции...

поскольку я использую RocksDB в качестве бэкэнда состояния для моей работы Flink, и я настраиваю параметры этого бэкэнда состояния, я прочитал на этой странице Flink , что у меня есть два варианта сохранения таймеров (RockDB или куча), и я прочитал...

Привет, я пытаюсь выполнить контрольную точку в одном из моих модулей flink, в котором я использую CoFlatMapFunction для объединения с потоками, если я закомментирую контрольную точку CoFlatMapFunction, работает, если раскомментировать снова, она не...