У меня есть два потока событий: один генерирует событие, сигнализирующее о начале жизни элемента, и другой поток, который генерирует событие, сигнализирующее об окончании срока жизни элемента. (Потоки могут быть объединены на itemId .) Как я могу...

Я использовал JMX в качестве репортера метрик, чтобы получить метрики Flink, но есть ли способ получить их в качестве вывода в терминале? Я хочу построить numRecordsInPerSecond для каждого оператора для анализа производительности, как я могу это...

Я подписался на этот пример и реализован с помощью kafka json с теми же примерами данных. выборочные данные потребителей {"temperature" : 28,"machineName":"xyz"} DataStream<Alert> patternStream = CEP.pattern(inputEventStream,...

Я использовал модуль Flink CEP и задавался вопросом, передаю ли я функцию в предложение where, которое будет возвращать логическое значение, будет ли оно работать распределенным образом или нет. Пример-: val pattern=...

Может ли кто-нибудь сообщить мне, как я могу настроить Flink на моей бессерверной платформе (FaaS) для выполнения операций, управляемых событиями? Я посмотрел на функции Flink, и они кажутся многообещающими. Может ли кто-нибудь уточнить...

Всем. Пожалуйста, помогите мне. Я пишу задание потоковой передачи apache flink, которое считывает json-сообщения из apache kafka (500-1000 сообщений в секундах), десериализует их в POJO и выполняет некоторые операции (фильтр-ключ-приемник-процесс)....

предположим, что у меня есть эта конфигурация для дескриптора, и действия были предприняты отсюда: ValueStateDescriptor<Event> descriptor = ...; StateTtlConfig ttlConfigOneHourAndReturnExpire = StateTtlConfig.newBuilder(Time.hours(1))...

Основная программа потребляет события кафки, затем фильтрует - ›карта -› keyBy - ›CEP -› сток. Я написал еще одну отдельную простую программу для чтения каталога контрольных точек, например: object StateReader extends App { val path =...