Flink 使用介绍相关文档目录
Job提交出现异常：No ExecutorFactory found to execute the application
详细的报错如下所示：
Exception in thread "main" java.lang.IllegalStateException: No ExecutorFactory found to execute the application.
at org.apache.flink.core.execution.DefaultExecutorServiceLoader.getExecutorFactory(DefaultExecutorServiceLoader.java:84)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1931)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1836)
at org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:70)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1822)
at com.paultech.CEPProblem.main(CEPProblem.java:73)
原因分析：
经过排查，发现是从Flink 1.11开始，flink-streaming-java中的flink-clients依赖被移除。运行的时候需要添加这个依赖，如下所示：
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
其中scala.binary.version和flink.version属性修改为实际项目中使用的版本
StackOverflow中相关问题链接：https://stackoverflow.com/questions/63032060/upgraded-flink-from-1-10-to-1-11-met-error-no-executorfactory-found-to-execute
Flink CEP 作业执行结果异常
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<TemperatureEvent> input = env.fromElements(new TemperatureEvent(1, "Device01", 22.0),
new TemperatureEvent(1, "Device01", 27.1), new TemperatureEvent(2, "Device01", 28.1),
new TemperatureEvent(1, "Device01", 22.2), new TemperatureEvent(3, "Device01", 22.1),
new TemperatureEvent(1, "Device02", 22.3), new TemperatureEvent(4, "Device02", 22.1),
new TemperatureEvent(1, "Device02", 22.4), new TemperatureEvent(5, "Device02", 22.7),
new TemperatureEvent(1, "Device02", 27.0), new TemperatureEvent(6, "Device02", 30.0));
Pattern<TemperatureEvent, ?> warningPattern = Pattern.<TemperatureEvent>begin("start")
.subtype(TemperatureEvent.class)
.where(new SimpleCondition<TemperatureEvent>() {
@Override
public boolean filter(TemperatureEvent subEvent) {
if (subEvent.getTemperature() >= 26.0) {
return true;
}
return false;
}
}).where(new SimpleCondition<TemperatureEvent>() {
@Override
public boolean filter(TemperatureEvent subEvent) {
if (subEvent.getMachineName().equals("Device02")) {
return true;
}
return false;
}
}).within(Time.seconds(10));
DataStream<Alert> patternStream = CEP.pattern(input, warningPattern)
.select(
new RichPatternSelectFunction<TemperatureEvent, Alert>() {
/**
*/
private static final long serialVersionUID = 1L;
@Override
public void open(Configuration parameters) throws Exception {
System.out.println(getRuntimeContext().getUserCodeClassLoader());
}
@Override
public Alert select(Map<String, List<TemperatureEvent>> event) throws Exception {
return new Alert("Temperature Rise Detected: " + event.get("start") + " on machine name: " + event.get("start"));
}
});
patternStream.print();
env.execute("CEP on Temperature Sensor");
TemperatureEvent 和 Alert类为Java bean，不再贴出它们的代码
这段代码在1.12版本之前执行会输出：
Alert{message='Temperature Rise Detected: [TemperatureEvent{id=1, machineName='Device02', temperature=27.0}] on machine name: [TemperatureEvent{id=1, machineName='Device02', temperature=27.0}]'}
Alert{message='Temperature Rise Detected: [TemperatureEvent{id=6, machineName='Device02', temperature=30.0}] on machine name: [TemperatureEvent{id=6, machineName='Device02', temperature=30.0}]'}
但是在Flink 1.12版本中运行，patternStream.print()这一行没有任何输出。debug了CepOperator一直到StreamOneInputProcessor，发现均没有元素输入，甚是奇怪。
第一反应可能是Time characteristic设置的问题。于是在创建env的后面添加了一行：
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
然后发现这个调用在Flink 1.12版本中被废弃了。原因是Flink 1.12中的流处理默认的时间特征改为TimeCharacteristic.EventTime，也就是说从之前默认的processing time改为了event time。这就是问题所在。对于默认的event time，Flink需要等待到10秒内的元素都到齐（下一个元素的event time为10秒后），才会打印出结果。然而，10秒钟之后的元素还没有到来时Flink 程序就退出了，因此不会有任何输出。
要解决这个问题，我们需要知道1.12版本中CEP设置时间特征为ProcessingTime的方式。尝试使用前边提到的配置方法无效。
如何在CEP处理流程中使用ProcessingTime呢？我们翻阅PatternStream的源代码，发现有如下两个方法：
/**
* Sets the time characteristic to processing time.
*/
public PatternStream<T> inProcessingTime() {
return new PatternStream<>(builder.inProcessingTime());
}
/**
* Sets the time characteristic to event time.
*/
public PatternStream<T> inEventTime() {
return new PatternStream<>(builder.inEventTime());
}
显然，这两个方法分别是用来设置ProcessingTime和EventTime的。将本篇开始的代码修改，如下所示：
DataStream<Alert> patternStream = CEP.pattern(input, warningPattern)
.inProcessingTime() // 注意，这里是关键，显式指定使用processing time
.select(
new RichPatternSelectFunction<TemperatureEvent, Alert>() {
/**
*/
private static final long serialVersionUID = 1L;
@Override
public void open(Configuration parameters) throws Exception {
System.out.println(getRuntimeContext().getUserCodeClassLoader());
}
@Override
public Alert select(Map<String, List<TemperatureEvent>> event) throws Exception {
return new Alert("Temperature Rise Detected: " + event.get("start") + " on machine name: " + event.get("start"));
}
});
再次运行，发现打印出了预期结果。
