多种事件类型的 Apache Flink CEP 模式
Apache Flink CEP pattern of multiple event types
目前我正在做一个学期项目,我必须识别三个事件的系列。喜欢P -> R -> P
我们有两种不同的事件类型,它们通过同一主题中的 Kafka 连接器使用。
我创建了一个名为 Event 的父 class,其他两种类型都从中派生。
Kafka 连接器使用 EventSchema 将 JSON 反序列化到父 class 事件。
val consumer = new FlinkKafkaConsumer("events", new EventSchema, properties)
val stream = env.addSource(consumer)
模式看起来像这样:
val pattern = Pattern
.begin[Event]("before")
.subtype(classOf[Position])
.next("recognized")
.subtype(classOf[Recognized])
.next("after")
.subtype(classOf[Position])
目前的问题是,如果我以适当的格式发送三条消息,则无法识别该模式。
我还试过什么..我改变了这样的模式:
val pattern = Pattern
.begin[Event]("before")
.where(e => e.getType == "position")
.next("recognized")
.where(e => e.getType == "recognition")
.next("after")
.where(e => e.getType == "position")
此模式有效,但后来我无法将事件 class 转换为定位或识别..
我在这里想念什么?
根据评论,我认为您应该 return 子类型实例而不是事件。这是我为您提供的示例代码:
val event = mapper.readValue(bytes, classOf[Event])
event.getType match {
case "position" => mapper.readValue(bytes, classOf[Position])
case "recognition" => mapper.readValue(bytes, classOf[Recognized])
case _ =>
}
我在 CEPITCase.java.
中成功尝试了测试用例中的示例
DataStream<Event> input = env.fromElements(
new Event(1, "foo", 4.0),
new SubEvent(2, "foo", 4.0, 1.0),
new SubEvent(3, "foo", 4.0, 1.0),
new SubEvent(4, "foo", 4.0, 1.0),
new Event(5, "middle", 5.0)
);
Pattern<Event, ?> pattern = Pattern.<Event>begin("start").subtype(SubEvent.class)
.followedByAny("middle").subtype(SubEvent.class)
.followedByAny("end").subtype(SubEvent.class);
目前我正在做一个学期项目,我必须识别三个事件的系列。喜欢P -> R -> P
我们有两种不同的事件类型,它们通过同一主题中的 Kafka 连接器使用。
我创建了一个名为 Event 的父 class,其他两种类型都从中派生。
Kafka 连接器使用 EventSchema 将 JSON 反序列化到父 class 事件。
val consumer = new FlinkKafkaConsumer("events", new EventSchema, properties)
val stream = env.addSource(consumer)
模式看起来像这样:
val pattern = Pattern
.begin[Event]("before")
.subtype(classOf[Position])
.next("recognized")
.subtype(classOf[Recognized])
.next("after")
.subtype(classOf[Position])
目前的问题是,如果我以适当的格式发送三条消息,则无法识别该模式。
我还试过什么..我改变了这样的模式:
val pattern = Pattern
.begin[Event]("before")
.where(e => e.getType == "position")
.next("recognized")
.where(e => e.getType == "recognition")
.next("after")
.where(e => e.getType == "position")
此模式有效,但后来我无法将事件 class 转换为定位或识别..
我在这里想念什么?
根据评论,我认为您应该 return 子类型实例而不是事件。这是我为您提供的示例代码:
val event = mapper.readValue(bytes, classOf[Event])
event.getType match {
case "position" => mapper.readValue(bytes, classOf[Position])
case "recognition" => mapper.readValue(bytes, classOf[Recognized])
case _ =>
}
我在 CEPITCase.java.
中成功尝试了测试用例中的示例DataStream<Event> input = env.fromElements(
new Event(1, "foo", 4.0),
new SubEvent(2, "foo", 4.0, 1.0),
new SubEvent(3, "foo", 4.0, 1.0),
new SubEvent(4, "foo", 4.0, 1.0),
new Event(5, "middle", 5.0)
);
Pattern<Event, ?> pattern = Pattern.<Event>begin("start").subtype(SubEvent.class)
.followedByAny("middle").subtype(SubEvent.class)
.followedByAny("end").subtype(SubEvent.class);