import org.apache.flink.cep.PatternSelectFunction; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

	Properties properties = new Properties();
	properties.setProperty("bootstrap.servers", "localhost:9092");
	properties.setProperty("group.id", "test");

	DataStream<TemperatureEvent> inputEventStream = env.addSource(
			new FlinkKafkaConsumer09<TemperatureEvent>("test", new EventDeserializationSchema(), properties));

	Pattern<TemperatureEvent, ?> warningPattern = Pattern.<TemperatureEvent> begin("first")
			.subtype(TemperatureEvent.class).where(new FilterFunction<TemperatureEvent>() {
				private static final long serialVersionUID = 1L;

				public boolean filter(TemperatureEvent value) {
					if (value.getTemperature() >= 26.0) {
						return true;
					}
					return false;
				}
			}).within(Time.seconds(10));

	DataStream<Alert> patternStream = CEP.pattern(inputEventStream, warningPattern)
			.select(new PatternSelectFunction<TemperatureEvent, Alert>() {
				private static final long serialVersionUID = 1L;

				public Alert select(Map<String, TemperatureEvent> event) throws Exception {

					return new Alert("Temperature Rise Detected:" + event.get("first").getTemperature()
							+ " on machine name:" + event.get("first").getMachineName());
				}

			});

	patternStream.print();
	env.execute("CEP on Temperature Sensor");
}
 

import org.apache.flink.cep.PatternSelectFunction; //导入依赖的package包/类
private static <TWarningType extends IWarning> DataStream<TWarningType> toWarningStream(DataStream<LocalWeatherData> localWeatherDataDataStream, IWarningPattern<LocalWeatherData, TWarningType> warningPattern) {
    PatternStream<LocalWeatherData> tempPatternStream = CEP.pattern(
            localWeatherDataDataStream.keyBy(new KeySelector<LocalWeatherData, String>() {
                @Override
                public String getKey(LocalWeatherData localWeatherData) throws Exception {
                    return localWeatherData.getStation().getWban();
                }
            }),
            warningPattern.getEventPattern());

    DataStream<TWarningType> warnings = tempPatternStream.select(new PatternSelectFunction<LocalWeatherData, TWarningType>() {
        @Override
        public TWarningType select(Map<String, List<LocalWeatherData>> map) throws Exception {
            return warningPattern.create(map);
        }
    }, new GenericTypeInfo<TWarningType>(warningPattern.getWarningTargetType()));

    return warnings;
}
 

import org.apache.flink.cep.PatternSelectFunction; //导入依赖的package包/类
public SelectTimeoutCepOperator(
	TypeSerializer<IN> inputSerializer,
	boolean isProcessingTime,
	NFACompiler.NFAFactory<IN> nfaFactory,
	final EventComparator<IN> comparator,
	AfterMatchSkipStrategy skipStrategy,
	PatternSelectFunction<IN, OUT1> flatSelectFunction,
	PatternTimeoutFunction<IN, OUT2> flatTimeoutFunction,
	OutputTag<OUT2> outputTag) {
	super(
		inputSerializer,
		isProcessingTime,
		nfaFactory,
		comparator,
		skipStrategy,
		new SelectWrapper<>(flatSelectFunction, flatTimeoutFunction));
	this.timedOutOutputTag = outputTag;
}
 

import org.apache.flink.cep.PatternSelectFunction; //导入依赖的package包/类
public static <K> SelectCepOperator<Event, K, Map<String, List<Event>>> getKeyedCepOpearator(
	boolean isProcessingTime,
	NFACompiler.NFAFactory<Event> nfaFactory,
	EventComparator<Event> comparator) {
	return new SelectCepOperator<>(
		Event.createTypeSerializer(),
		isProcessingTime,
		nfaFactory,
		comparator,
		null,
		new PatternSelectFunction<Event, Map<String, List<Event>>>() {
			@Override
			public Map<String, List<Event>> select(Map<String, List<Event>> pattern) throws Exception {
				return pattern;
			}
		});
}
 

import org.apache.flink.cep.PatternSelectFunction; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	DataStream<TemperatureEvent> inputEventStream = env.fromElements(new TemperatureEvent("xyz", 22.0),
			new TemperatureEvent("xyz", 20.1), new TemperatureEvent("xyz", 21.1), new TemperatureEvent("xyz", 22.2),
			new TemperatureEvent("xyz", 29.1), new TemperatureEvent("xyz", 22.3), new TemperatureEvent("xyz", 22.1),
			new TemperatureEvent("xyz", 22.4), new TemperatureEvent("xyz", 22.7),
			new TemperatureEvent("xyz", 27.0));

	Pattern<TemperatureEvent, ?> warningPattern = Pattern.<TemperatureEvent> begin("first")
			.subtype(TemperatureEvent.class).where(new FilterFunction<TemperatureEvent>() {
				private static final long serialVersionUID = 1L;

				public boolean filter(TemperatureEvent value) {
					if (value.getTemperature() >= 26.0) {
						return true;
					}
					return false;
				}
			}).within(Time.seconds(10));

	DataStream<Alert> patternStream = CEP.pattern(inputEventStream, warningPattern)
			.select(new PatternSelectFunction<TemperatureEvent, Alert>() {
				private static final long serialVersionUID = 1L;

				public Alert select(Map<String, TemperatureEvent> event) throws Exception {

					return new Alert("Temperature Rise Detected");
				}

			});

	patternStream.print();
	env.execute("CEP on Temperature Sensor");
}
 

import org.apache.flink.cep.PatternSelectFunction; //导入依赖的package包/类
public SelectCepOperator(
	TypeSerializer<IN> inputSerializer,
	boolean isProcessingTime,
	NFACompiler.NFAFactory<IN> nfaFactory,
	EventComparator<IN> comparator,
	AfterMatchSkipStrategy skipStrategy,
	PatternSelectFunction<IN, OUT> function) {
	super(inputSerializer, isProcessingTime, nfaFactory, comparator, skipStrategy, function);
}
 

import org.apache.flink.cep.PatternSelectFunction; //导入依赖的package包/类
PatternSelectFunction<IN, OUT1> getFlatSelectFunction() {
	return flatSelectFunction;
}
 

import org.apache.flink.cep.PatternSelectFunction; //导入依赖的package包/类
public SelectWrapper(
	PatternSelectFunction<IN, OUT1> flatSelectFunction,
	PatternTimeoutFunction<IN, OUT2> flatTimeoutFunction) {
	this.flatSelectFunction = flatSelectFunction;
	this.flatTimeoutFunction = flatTimeoutFunction;
}