ApacheFlink-CEP示例的简单ScalaAPI不显示任何输出


0

我正在编写一个简单的示例,用最新的Github版本1.1-SNAPSHOT testFlink中CEP的新scalaapi。

模式只是对值的检查,并为每个匹配的模式输出一个字符串作为结果。代码如下:

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输出没有显示print()的迹象。即使放松模式条件,只设置一个“开始”(接受所有事件)也绝对不会返回任何结果。

另外,当尝试添加阶段时,代码无法编译。如果我将模式更改为(查找第三个字段小于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))

在“start”阶段之后的第一个where()上显示错误。我试图显式设置参数类型:

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

这样它会再次编译,但是当它在Flink上运行时,就不会显示任何输出。StreamingAlert是一个Scala数据流[(String,Long,Int)],在代码的其他部分,我可以毫无问题地使用.4进行过滤,并且输出看起来是正确的。

1 答案


0

流式API中的print()API调用不会触发立即执行。你还得 调用环境执行()在你的课程结束时。

定义模式时,应该在某个地方提供事件类型。要么按原样执行,要么通过begin的类型参数执行:

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

我来回答

写文章

提问题

面试题