Простой API Scala для примера CEP не показывает никакого вывода

Question

Я программирую простой пример для тестирования нового Scala API для CEP в Flink, используя последнюю версию Github для 1.1-SNAPSHOT.

Шаблон - это только проверка значения, и в результате получается один String для каждого сопоставленного шаблона. Код выглядит следующим образом:

val pattern : Pattern[(String, Long, Int), _] = Pattern.begin("start").where(_._3 < 4) 

val cepEventAlert = CEP.pattern(streamingAlert, pattern) 

def selectFn(pattern : mutable.Map[String, (String, Long, Int)]): String = { 
    val startEvent = pattern.get("start").get 
    "Alerta:"+startEvent._1+": Pattern" 
} 

val patternStreamSelected = cepEventAlert.select(selectFn(_)) 

patternStreamSelected.print() 

Он собирает и работает под 1.1-SNAPSHOT без проблем, но выход JobManager не показывает никаких признаков, что печать(). Даже расслабление условий шаблона и установка только «начала» (принятие всех событий) абсолютно ничего не возвращает.

Кроме того, при попытке добавить этапы код не удается скомпилировать. Если изменить шаблон на (Обнаружение двух последовательных событий с 3-го поля меньше, чем 4):

Pattern.begin("start").where(_._3 < 4).next("end").where(_._3 < 4).within(Time.seconds(30)) 

Компилятор затем Броски:

error: missing parameter type for expanded function ((x$4) => x$4._3.$less(4)) 

Отображение ошибки на первой, где() после того, как «старт». Я судимый явно установить тип параметра с помощью:

(x: (String, Long, Int)) => x._3 < 4 

Таким образом, он собирает снова, но когда он работает на Флинка, то вывод не показан. StreamingAlert - это Scala DataStream [(String, Long, Int)], и в других частях кода я могу без проблем фильтровать с _._ < 4, и результат кажется правильным.

Answer 1

Ошибка API в потоковом API не вызывает нетерпеливого исполнения. Вам еще нужно позвонить env.execute() в конце вашей программы.

Когда вы определяете свой шаблон, вы должны указать тип события где-нибудь. Либо вы делаете это, как вы сделали это или сделать это с помощью параметра типа для begin:

Pattern.begin[(String, Long, Int)]("start").where(_._3 < 4).next("end").where(_._3 < 4).within(Time.seconds(30))