我正在尝试用cep-flink库开发一个简单的监控过程。我生产一个 csv Kafka的文件，我和Flink一起使用它，它将是一个tuple4。
在这里之前一切正常。flink consumer可以完美地使用来自kafka的数据并将其转换为tuple4。然后我为datastream对象创建pattern、patternstream和select（）函数。
显然一切正常，但不起作用。select（）函数有问题。
我还创建了一个pojo来接收最终值。我加上两个代码。谢谢。

public class LTMonitoringTest {
private static final int LEUKO = 3500;

public static void main(final String[] ARGS) throws Exception{
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    env.enableCheckpointing(1000);

    ParameterTool parameterTool = ParameterTool.fromArgs(ARGS);

    DataStreamSource<String> myConsumer = env.addSource(new FlinkKafkaConsumer082<>
            (parameterTool.getRequired("topic"), new SimpleStringSchema(),
                    parameterTool.getProperties()));

    DataStream<Tuple4<Integer, Integer, Integer, Long>> streamTuple = myConsumer
            .map(new csv2Tuple())
            .assignTimestampsAndWatermarks(new ExtractTimestamp());

    streamTuple.print();

    Pattern<Tuple4<Integer, Integer, Integer, Long>, ?> warningPattern =
            Pattern.<Tuple4<Integer, Integer, Integer, Long>>
            begin("first")
            .where(new IterativeCondition<Tuple4<Integer, Integer, Integer, Long>>() {
                @Override
                public boolean filter(Tuple4<Integer, Integer, Integer, Long> value,
                                      Context<Tuple4<Integer, Integer, Integer, Long>> ctx)
                        throws Exception {
                    return value.f1 <= LEUKO;
                }
            });

    PatternStream<Tuple4<Integer, Integer, Integer, Long>> tempPatternStream = CEP.pattern(
            streamTuple.keyBy(0),
            warningPattern);

    DataStream<Warning> warnings1 = tempPatternStream.select(
            (Map<String, List<Tuple4<Integer, Integer, Integer, Long>>> pattern) -> {
                Tuple4<Integer, Integer, Integer, Long> first =
                        (Tuple4<Integer, Integer, Integer, Long>) pattern.get("first").get(0);

                return new Warning(first.f0, first.f1);
            }
    );

    warnings1.print();
    env.execute("CEP monitoring job");
}

private static class ExtractTimestamp extends
        AscendingTimestampExtractor<Tuple4<Integer, Integer, Integer, Long>> {

    @Override
    public long extractAscendingTimestamp(Tuple4<Integer, Integer, Integer, Long> element) {
        return element.f3;
    }
}

public static class csv2Tuple implements MapFunction<String,
        Tuple4<Integer, Integer, Integer, Long>> {
    @Override
    public Tuple4<Integer, Integer, Integer, Long> map(String str) {
        String[] temp = str.split(",");
        return new Tuple4<>(
                Integer.parseInt(temp[0]),
                Integer.parseInt(temp[1]),
                Integer.parseInt(temp[2]),
                Long.parseLong(temp[3]));

    }

}

}
它作为参数接收（对于kafka konsumer）：
--主题lt监控--bootstrap.serverslocalhost:9092 --zookeeper.connect localhost:2181 --group.id mygroup公司
还有pojo：

public class Warning {

private int ID;
private int Level;

public Warning(int ID, int Level) {
    this.ID = ID;
    this.Level = Level;
}

/**
 * Sets new ID.
 *
 * @param ID New value of ID.
 */
public void setID(int ID) {
    this.ID = ID;
}

/**
 * Gets ID.
 *
 * @return Value of ID.
 */
public int getID() {
    return ID;
}

/**
 * Gets Level.
 *
 * @return Value of Level.
 */
public int getLevel() {
    return Level;
}

/**
 * Sets new Level.
 *
 * @param Level New value of Level.
 */
public void setLevel(int Level) {
    this.Level = Level;
}

@Override
public boolean equals(Object obj) {
    if (obj instanceof Warning) {
        Warning other = (Warning) obj;

        return ID == other.ID && Level == other.Level;
    } else {
        return false;
    }
}

@Override
public int hashCode() {
    return 41 * ID + Double.hashCode(Level);
}

@Override
public String toString() {
    return "Warning{" +
            "ID=" + ID + getID() +
            ", Level=" + Level +
            '}';
}

}
所以请想想为什么它不运行？
顺便说一下，这是一个错误：
线程“main”java.lang.nosuchmethoderror中出现异常：org.apache.flink.api.java.typeutils.typeextractor.getunaryoperatorreturntype（lorg/apache/flink/api/common/functions/function；ljava/lang/class；ii[i[ilorg/apache/flink/api/common/typeinfo/typeinformation；ljava/lang/string；z） lorg/apache/flink/api/common/typeinfo/typeinformation；在org.apache.flink.cep.patternstream.select（patternstream。java:96)在ltmonitoringtest.main（ltmonitoringtest。java:78)