我试图为kafka输入流实现一个非常简单的apache-flink-cep。Kafka生产者生成一个简单的双值,并通过Kafka主题将它们作为字符串发送给消费者。目前我正在用flink编写cep消费者代码。到目前为止,这是我写的代码:
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().disableSysoutLogging();
env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(4, 10000));
env.setParallelism(3);
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "flink_consumer");
DataStream<String> stream = env
.addSource(new FlinkKafkaConsumer09<>("temp", new SimpleStringSchema(), properties));
Pattern<String, ?> warning= Pattern.<String>begin("first")
.where(new IterativeCondition<String>() {
private static final long serialVersionUID = 1L;
@Override
public boolean filter(String value, Context<String> ctx) throws Exception {
return Double.parseDouble(value) >= 89.0;
}
})
.next("second")
.where(new IterativeCondition<String>() {
private static final long serialVersionUID = 1L;
@Override
public boolean filter(String value, Context<String> ctx) throws Exception {
return Double.parseDouble(value) >= 89.0;
}
})
.within(Time.seconds(10));
DataStream<String> temp = CEP.pattern(stream, warning).select(new PatternSelectFunction<String, String>() {
private static final long serialVersionUID = 1L;
@Override
public String select(Map<String, List<String>> pattern) throws Exception {
List warnung1 = pattern.get("first");
String first = (String) warnung1.get(1);
return first;
}
});
temp.print();
env.execute();
}
如果我试图执行此代码,则会显示以下错误消息:
线程“main”java.lang.nosuchfielderror中出现异常:org.apache.flink.cep.patternstream.select(patternstream)中没有索引。java:102)在ceptest.main(ceptest。java:50)
所以看起来我用cep模式生成的数据流是错误的,但是我不知道这个方法有什么问题。太好了!
编辑:我尝试了一些其他的例子,每次执行我都会得到相同的错误。所以我觉得我的包裹有问题?
1条答案
按热度按时间c3frrgcw1#
使用flink1.6.0,我的代码可以完美地工作。