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