Я пытаюсь реализовать очень простой Apache Flink CEP для Kafka InputStream. Производитель Kafka генерирует простые двойные значения и отправляет их через тему Kafka в виде строки потребителям. В настоящий момент я кодирую CEP Consumer с помощью Flink. Пока это мой написанный код:
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: NO_INDEX в org.apache.flink.cep.PatternStream.select (PatternStream.java:102) в CEPTest.main (CEPTest.java:50)
Итак, похоже, что мой сгенерированный DataStream с шаблоном CEP неверен, но я не знаю, что не так с этим методом. Любая помощь была бы замечательной!
Изменить: я пробовал другой пример, и при каждом выполнении я получаю ту же ошибку. Я думаю, что с моими пакетами что-то не так?
