stopWithSavepoint
org.apache.flink.client.program.rest.RestClusterClient#stopWithSavepoint
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| public CompletableFuture<String> stopWithSavepoint(JobID jobId,
boolean advanceToEndOfTime,
@Nullable
String savepointDirectory)
Description copied from interface: ClusterClient
停止在 flink 集群上的一个程序，该程序的作业管理器配置在该客户端的配置中。停止只对流任务有效。
请注意，在发送停止命令之后，程序可能会继续运行一段时间，因为在源停止发出数据之后，所有操作员都需要完成处理.
Specified by:
stopWithSavepoint in interface ClusterClient<T>
Parameters:
jobId - 要停止的流任务的作业ID
advanceToEndOfTime - 源是否应在管道中注入 max_watermark 的标志
savepointDirectory - 保存点应该写入的目录
Returns:
CompletableFuture 包含保存点所在的路径
|
在处理Operator-state时，clear逻辑可能会出问题:
如何指定了advanceToEndOfTime=true
在CustomProcessWindowFunction中定义了MapState，在clear方法中指定了清理state的逻辑时，需要判断当前时间戳是否为 max_watermark .
下发max_watermark，会触发关闭掉所有窗口。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| eventItemStream.keyBy("tag")
.timeWindow(Time.minutes(1))
.trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(10)))
.evictor(TimeEvictor.of(Time.seconds(0), true))
.allowedLateness(Time.seconds(0))
.process(new CustomProcessWindowFunction())
.print("data: ");
@Override
public void clear(ProcessWindowFunction<EventItem, EventItem, Tuple, TimeWindow>.Context context) throws Exception {
super.clear(context);
if(context.currentWatermark() != Long.MAX_VALUE){
log.error("do clear logic!");
}
}
|