Flink1.13架构全集| 一文带你由浅入深精通Flink方方面面（二）

public abstract class ProcessFunction<I, O> extends AbstractRichFunction {
    ...
public abstract void processElement(I value, Context ctx, Collector<O> out) throws Exception;
public void onTimer(long timestamp, OnTimerContext ctx, Collector<O> out) throws Exception {}
...
}

value：当前流中的输入元素，也就是正在处理的数据，类型与流中数据类型一致。
ctx：类型是ProcessFunction中定义的内部抽象类Context，表示当前运行的上下文，可以获取到当前的时间戳，并提供了用于查询时间和注册定时器的“定时服务”(TimerService)，以及可以将数据发送到“侧输出流”（side output）的方法.output()。
out：“收集器”（类型为Collector），用于返回输出数据。使用方式与flatMap算子中的收集器完全一样，直接调用out.collect()方法就可以向下游发出一个数据。这个方法可以多次调用，也可以不调用。
通过几个参数的分析不难发现，ProcessFunction可以轻松实现flatMap这样的基本转换功能（当然map、filter更不在话下）；而通过富函数提供的获取上下文方法.getRuntimeContext()，也可以自定义状态（state）进行处理，这也就能实现聚合操作的功能了。

（1）ProcessFunction
最基本的处理函数，基于DataStream直接调用.process()时作为参数传入。
（2）KeyedProcessFunction
对流按键分区后的处理函数，基于KeyedStream调用.process()时作为参数传入。要想使用定时器，比如基于KeyedStream。
（3）ProcessWindowFunction
开窗之后的处理函数，也是全窗口函数的代表。基于WindowedStream调用.process()时作为参数传入。
（4）ProcessAllWindowFunction
同样是开窗之后的处理函数，基于AllWindowedStream调用.process()时作为参数传入。
（5）CoProcessFunction
合并（connect）两条流之后的处理函数，基于ConnectedStreams调用.process()时作为参数传入。关于流的连接合并操作，我们会在后续章节详细介绍。
（6）ProcessJoinFunction
间隔连接（interval join）两条流之后的处理函数，基于IntervalJoined调用.process()时作为参数传入。
（7）BroadcastProcessFunction
广播连接流处理函数，基于BroadcastConnectedStream调用.process()时作为参数传入。这里的“广播连接流”BroadcastConnectedStream，是一个未keyBy的普通DataStream与一个广播流（BroadcastStream）做连接（conncet）之后的产物。关于广播流的相关操作，我们会在后续章节详细介绍。
（8）KeyedBroadcastProcessFunction
按键分区的广播连接流处理函数，同样是基于BroadcastConnectedStream调用.process()时作为参数传入。与BroadcastProcessFunction不同的是，这时的广播连接流，是一个KeyedStream与广播流（BroadcastStream）做连接之后的产物。
接下来，我们就对KeyedProcessFunction和ProcessWindowFunction的具体用法展开详细说明。
11.2 按键分区处理函数（KeyedProcessFunction）
在上节中提到，只有在KeyedStream中才支持使用TimerService设置定时器的操作。所以一般情况下，我们都是先做了keyBy分区之后，再去定义处理操作；代码中更加常见的处理函数是KeyedProcessFunction。

// 获取当前的处理时间
long currentProcessingTime();

// 获取当前的水位线（事件时间）
long currentWatermark();

// 注册处理时间定时器，当处理时间超过time时触发
void registerProcessingTimeTimer(long time);

// 注册事件时间定时器，当水位线超过time时触发
void registerEventTimeTimer(long time);

// 删除触发时间为time的处理时间定时器
void deleteProcessingTimeTimer(long time);

// 删除触发时间为time的处理时间定时器
void deleteEventTimeTimer(long time);

stream.keyBy( t -> t.f0 )
.process(new MyKeyedProcessFunction())

public class ProcessingTimeTimerExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new CustomSource())
                .keyBy(r -> true)
                .process(new KeyedProcessFunction<Boolean, String, String>() {
                    // 每来一条数据都会调用一次
                    @Override
                    public void processElement(String s, Context context, Collector<String> collector) throws Exception {
                        long currTs = context.timerService().currentProcessingTime();
                        collector.collect("数据到达，到达时间是：" + new Timestamp(currTs));
                        // 注册10s之后的定时器
                        context.timerService().registerProcessingTimeTimer(currTs + 10 * 1000L);
                    }

                    // 定时器触发时的操作
                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        out.collect("定时器触发，触发时间是：" + new Timestamp(timestamp));
                    }
                })
                .print();

        env.execute();
    }

    public static class CustomSource implements SourceFunction<String> {
        @Override
        public void run(SourceFunction.SourceContext<String> ctx) throws Exception {
            ctx.collect("a");
            // 为了让程序不直接退出，等待20秒
            Thread.sleep(20 * 1000L);
        }

        @Override
        public void cancel() {
        }
    }
}

stream.keyBy( t -> t.f0 )
        .window( TumblingEventTimeWindows.of(Time.seconds(10)) )
        .process(new MyProcessWindowFunction())

public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window>
        extends AbstractRichFunction {
...
public abstract void process(
        KEY key, Context context, Iterable<IN> elements, Collector<OUT> out) throws Exception;
public void clear(Context context) throws Exception {}
public abstract class Context implements java.io.Serializable {...}
}

IN：input，数据流中窗口任务的输入数据类型。
OUT：output，窗口任务进行计算之后的输出数据类型。
KEY：数据中键key的类型。
W：窗口的类型，是Window的子类型。一般情况下我们定义时间窗口，W就是TimeWindow。
而内部定义的方法，跟我们之前熟悉的处理函数就有所区别了。因为全窗口函数不是逐个处理元素的，所以处理数据的方法在这里并不是.processElement()，而是改成了.process()。方法包含四个参数。
key：窗口做统计计算基于的键，也就是之前keyBy用来分区的字段。
context：当前窗口进行计算的上下文，它的类型就是ProcessWindowFunction内部定义的抽象类Context。
elements：窗口收集到用来计算的所有数据，这是一个可迭代的集合类型。
out：用来发送数据输出计算结果的收集器，类型为Collector。
可以明显看出，这里的参数不再是一个输入数据，而是窗口中所有数据的集合。而上下文context所包含的内容也跟其他处理函数有所差别：

public abstract class Context implements java.io.Serializable {
    public abstract W window();

    public abstract long currentProcessingTime();
    public abstract long currentWatermark();

    public abstract KeyedStateStore windowState();
    public abstract KeyedStateStore globalState();
    public abstract <X> void output(OutputTag<X> outputTag, X value);
}

stream.windowAll( TumblingEventTimeWindows.of(Time.seconds(10)) )
.process(new MyProcessAllWindowFunction())

public class ProcessAllWindowTopN {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> pvStream = env
                .addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );

        pvStream
                .map(new MapFunction<Event, String>() {
                    @Override
                    public String map(Event value) throws Exception {
                        return value.url;
                    }
                })    // 只需要url，提取转换成String
                .windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))    // 开滑动窗口
                .process(new ProcessAllWindowFunction<String, String, TimeWindow>() {
                    @Override
                    public void process(Context context, Iterable<String> elements, Collector<String> out) throws Exception {
                        HashMap<String, Long> urlCountMap = new HashMap<>();
                        // 遍历窗口中数据，将浏览量保存到HashMap中
                        for (String url : elements) {
                            if (urlCountMap.containsKey(url)) {
                                long count = urlCountMap.get(url);
                                urlCountMap.put(url, count + 1L);
                            } else {
                                urlCountMap.put(url, 1L);
                            }
                        }
                        ArrayList<Tuple2<String, Long>> mapList = new ArrayList<Tuple2<String, Long>>();
                        // 将浏览量数据放入ArrayList，进行排序
                        for (String key : urlCountMap.keySet()) {
                            mapList.add(Tuple2.of(key, urlCountMap.get(key)));
                        }
                        mapList.sort(new Comparator<Tuple2<String, Long>>() {
                            @Override
                            public int compare(Tuple2<String, Long> o1, Tuple2<String, Long> o2) {
                                return o2.f1.intValue() - o1.f1.intValue();
                            }
                        });
                        StringBuilder result = new StringBuilder();
                        result.append("========================================\n");
                        // 取前两名，构建输出结果
                        for (int i = 0; i < 2; i++) {
                            Tuple2<String, Long> temp = mapList.get(i);
                            result
                                    .append("浏览量No." + (i + 1) + " ")
                                    .append("url：" + temp.f0 + " ")
                                    .append("浏览量：" + temp.f1 + " ")
                                    .append("窗口结束时间：" + new Timestamp(context.window().getEnd()) + "\n");
                        }
                        result
                                .append("========================================\n\n\n");
                        out.collect(result.toString());
                    }
                })
                .print();
        env.execute();
    }
}

总结处理流程如下：
（1）读取数据源；
（2）筛选浏览行为（pv）；
（3）提取时间戳并生成水位线；
（4）按照url进行keyBy分区操作；
（5）开长度为1小时、步长为5分钟的事件时间滑动窗口；
（6）使用增量聚合函数AggregateFunction，并结合全窗口函数WindowFunction进行窗口聚合，得到每个url、在每个统计窗口内的浏览量，包装成UrlViewCount；
（7）按照窗口进行keyBy分区操作；
（8）对同一窗口的统计结果数据，使用KeyedProcessFunction进行收集并排序输出。
用KeyedProcessFunction来收集数据做排序，这时面对的是窗口聚合之后的数据流，而窗口已经不存在了；我们需要确保能够收集齐所有数据，所以应该在窗口结束时间基础上再“多等一会儿”。具体实现上，可以采用一个延迟触发的事件时间定时器。基于窗口的结束时间来设定延迟，其实并不需要等太久——因为我们是靠水位线的推进来触发定时器，而水位线的含义就是“之前的数据都到齐了”。所以我们只需要设置1毫秒的延迟，就一定可以保证这一点。而在等待过程中，之前已经到达的数据应该缓存起来，我们这里用一个自定义的“列表状态”（ListState）来进行存储，如图所示。这个状态需要使用富函数类的getRuntimeContext()方法获取运行时上下文来定义，我们一般把它放在open()生命周期方法中。之后每来一个UrlViewCount，就把它添加到当前的列表状态中，并注册一个触发时间为窗口结束时间加1毫秒（windowEnd + 1）的定时器。待到水位线到达这个时间，定时器触发，我们可以保证当前窗口所有url的统计结果UrlViewCount都到齐了；于是从状态中取出进行排序输出。

public class KeyedProcessTopN {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> pvStream = env
                .addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );

        // 先求出每个url在每个窗口的浏览量
        SingleOutputStreamOperator<UrlViewCount> uvcStream = pvStream
                .keyBy(r -> r.url)
                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .aggregate(new CountAgg(), new WindowResult());    // 增量聚合，并结合全窗口函数包装UrlViewCount

        // 针对同一个窗口中的不同url的UrlViewCount次数，进行排序输出
        KeyedStream<UrlViewCount, Long> uvcKeyedStream = uvcStream
                .keyBy(r -> r.windowEnd);

        uvcKeyedStream
                .process(new TopN(2))
                .print();

        env.execute();
    }

    public static class TopN extends KeyedProcessFunction<Long, UrlViewCount, String> {
        // 列表状态变量
        private ListState<UrlViewCount> UrlViewCountListState;
        private Integer threshold;

        public TopN(Integer threshold) {
            this.threshold = threshold;
        }
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            // 声明一个列表状态，保存已经到达的统计结果
            UrlViewCountListState = getRuntimeContext().getListState(
                    new ListStateDescriptor<UrlViewCount>("list-state", Types.POJO(UrlViewCount.class))
            );
        }

        @Override
        public void processElement(UrlViewCount value, Context ctx, Collector<String> out) throws Exception {
            // 添加到列表状态变量中
            UrlViewCountListState.add(value);
            // 水位线达到 窗口结束时间 + 1毫秒 时触发定时器来进行排序
            ctx.timerService().registerEventTimeTimer(value.windowEnd + 1);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            // 将数据从列表状态变量中取出，放入ArrayList
            ArrayList<UrlViewCount> UrlViewCountArrayList = new ArrayList<>();
            for (UrlViewCount uvc : UrlViewCountListState.get()) {
                UrlViewCountArrayList.add(uvc);
            }
            // 清空状态释放资源
            UrlViewCountListState.clear();

            // 排序
            UrlViewCountArrayList.sort(new Comparator<UrlViewCount>() {
                @Override
                public int compare(UrlViewCount o1, UrlViewCount o2) {
                    return o2.count.intValue() - o1.count.intValue();
                }
            });

            // 取前三名，构建输出结果
            StringBuilder result = new StringBuilder();
            result.append("========================================\n");
            for (int i = 0; i < this.threshold; i++) {
                UrlViewCount UrlViewCount = UrlViewCountArrayList.get(i);
                result
                        .append("浏览量No." + (i + 1) + " ")
                        .append("url：" + UrlViewCount.url + " ")
                        .append("浏览量：" + UrlViewCount.count + " ")
                        .append("窗口结束时间：" + new Timestamp(timestamp - 1) + "\n");
            }
            result
                    .append("========================================\n\n\n");
            out.collect(result.toString());
        }
    }

    public static class CountAgg implements AggregateFunction<Event, Long, Long> {
        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(Event value, Long accumulator) {
            return accumulator + 1L;
        }

        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            return null;
        }
    }

    public static class WindowResult extends ProcessWindowFunction<Long, UrlViewCount, String, TimeWindow> {
        @Override
        public void process(String s, Context context, Iterable<Long> elements, Collector<UrlViewCount> out) throws Exception {
            out.collect(new UrlViewCount(s, elements.iterator().next(), context.window().getStart(), context.window().getEnd()));
        }
    }

	public static class UrlViewCount {
		public String url;
		public Long count;
		public Long windowStart;
		public Long windowEnd;
	
		public UrlViewCount() {
		}
	
		public UrlViewCount(String url, Long count, Long windowStart, Long windowEnd) {
			this.url = url;
			this.count = count;
			this.windowStart = windowStart;
			this.windowEnd = windowEnd;
		}
	
		@Override
		public String toString() {
			return "UrlViewCount{" +
					"url='" + url + '\'' +
					", count=" + count +
					", windowStart=" + new Timestamp(windowStart) +
					", windowEnd=" + new Timestamp(windowEnd) +
					'}';
		}
	}
}

private ListState<Event> UrlViewCountListState;

DataStream<Integer> stream = env.addSource(...);
SingleOutputStreamOperator<Long> longStream = stream.process(new ProcessFunction<Integer, Long>() {
      @Override
      public void processElement( Integer value, Context ctx, Collector<Integer> out) throws Exception {
        // 转换成Long，输出到主流中
        out.collect(Long.valueOf(value));
        // 转换成String，输出到侧输出流中
        ctx.output(outputTag, "side-output: " + String.valueOf(value));
      }
});

OutputTag<String> outputTag = new OutputTag<String>("side-output") {};

DataStream<String> stringStream = longStream.getSideOutput(outputTag);

public class SplitStreamByFilter {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<Event> stream = env
                .addSource(new ClickSource());
        // 筛选Mary的浏览行为放入MaryStream流中
        DataStream<Event> MaryStream = stream.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event value) throws Exception {
                return value.user.equals("Mary");
            }
        });
        // 筛选Bob的购买行为放入BobStream流中
        DataStream<Event> BobStream = stream.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event value) throws Exception {
                return value.user.equals("Bob");
            }
        });
        // 筛选其他人的浏览行为放入elseStream流中
        DataStream<Event> elseStream = stream.filter(new FilterFunction<Event>() {
            @Override
            public boolean filter(Event value) throws Exception {
                return !value.user.equals("Mary") && !value.user.equals("Bob") ;
            }
        });

        MaryStream.print("Mary pv");
        BobStream.print("Bob pv");
        elseStream.print("else pv");

        env.execute();

    }

    // 这里重新实现了ClickSource()，因为我们在自定义数据源中发送了水位线
    public static class ClickSource implements SourceFunction<Event> {
        private boolean running = true;
        @Override
        public void run(SourceContext<Event> sourceContext) throws Exception {
            Random random = new Random();
            String[] userArr = {"Mary", "Bob", "Alice"};
            String[] urlArr  = {"./home", "./cart", "./prod?id=1"};
            while (running) {
                long currTs = Calendar.getInstance().getTimeInMillis(); // 毫秒时间戳
                String username = userArr[random.nextInt(userArr.length)];
                String url      = urlArr[random.nextInt(urlArr.length)];
                Event event = new Event(username, url, currTs);
                // 使用collectWithTimestamp方法将数据发送出去，并指明数据中的时间戳的字段
                sourceContext.collectWithTimestamp(event, event.timestamp);
                // 发送水位线
                sourceContext.emitWatermark(new Watermark(event.timestamp - 1L));
                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}

public class SplitStreamByOutputTag {
    // 定义输出标签，侧输出流的数据类型为三元组(user, url, timestamp)
    private static OutputTag<Tuple3<String, String, Long>> MaryTag = new OutputTag<Tuple3<String, String, Long>>("Mary-pv"){};
    private static OutputTag<Tuple3<String, String, Long>> BobTag = new OutputTag<Tuple3<String, String, Long>>("Bob-pv"){};
    
public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<Event> stream = env
                .addSource(new ClickSource());

        SingleOutputStreamOperator<Event> processedStream = stream.process(new ProcessFunction<Event, Event>() {
            @Override
            public void processElement(Event value, Context ctx, Collector<Event> out) throws Exception {
                if (value.user.equals("Mary")){
                    ctx.output(MaryTag, new Tuple3<>(value.user, value.url, value.timestamp));
                } else if (value.user.equals("Bob")){
                    ctx.output(BobTag, new Tuple3<>(value.user, value.url, value.timestamp));
                } else {
                    out.collect(value);
                }
            }
        });

        processedStream.getSideOutput(MaryTag).print("Mary pv");
        processedStream.getSideOutput(BobTag).print("Bob pv");
        processedStream.print("else");

        env.execute();
    }

    // 这里重新实现了ClickSource()，因为我们在自定义数据源中发送了水位线
    public static class ClickSource implements SourceFunction<Event> {
        private boolean running = true;
        @Override
        public void run(SourceContext<Event> sourceContext) throws Exception {
            Random random = new Random();
            String[] userArr = {"Mary", "Bob", "Alice"};
            String[] urlArr  = {"./home", "./cart", "./prod?id=1"};
            while (running) {
                long currTs = Calendar.getInstance().getTimeInMillis(); // 毫秒时间戳
                String username = userArr[random.nextInt(userArr.length)];
                String url      = urlArr[random.nextInt(urlArr.length)];
                Event event = new Event(username, url, currTs);
                // 使用collectWithTimestamp方法将数据发送出去，并指明数据中的时间戳的字段
                sourceContext.collectWithTimestamp(event, event.timestamp);
                // 发送水位线
                sourceContext.emitWatermark(new Watermark(event.timestamp - 1L));
                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}

stream1.union(stream2, stream3, ...)

public class UnionExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Tuple2<String, Long>> stream1 = env
                .socketTextStream("localhost", 8888)
                .map(new MapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String value) throws Exception {
                        String[] arr = value.split(" ");
                        return Tuple2.of(arr[0], Long.parseLong(arr[1]) * 1000L);
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Tuple2<String, Long>>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                                        return element.f1;
                                    }
                                })
                );

        SingleOutputStreamOperator<Tuple2<String, Long>> stream2 = env
                .socketTextStream("localhost", 9999)
                .map(new MapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String value) throws Exception {
                        String[] arr = value.split(" ");
                        return Tuple2.of(arr[0], Long.parseLong(arr[1]) * 1000L);
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Tuple2<String, Long>>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                                        return element.f1;
                                    }
                                })
                );

        // 合并两条流进行处理
        stream1
                .union(stream2)
                .process(new ProcessFunction<Tuple2<String, Long>, String>() {
                    @Override
                    public void processElement(Tuple2<String, Long> in, Context context, Collector<String> collector) throws Exception {
                        collector.collect("当前水位线是：" + context.timerService().currentWatermark());
                    }
                })
                .print();

        env.execute();
    }
}

public class CoMapExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<Integer> stream1 = env.fromElements(1,2,3);
        DataStream<Long> stream2 = env.fromElements(1L,2L,3L);

        ConnectedStreams<Integer, Long> connectedStreams = stream1.connect(stream2);
        SingleOutputStreamOperator<String> result = connectedStreams.map(new CoMapFunction<Integer, Long, String>() {
            @Override
            public String map1(Integer value) {
                return "Integer: " + value;
            }

            @Override
            public String map2(Long value) {
                return "Long: " + value;
            }
        });

        result.print();

        env.execute();
    }
}

// 实时对账
public class BillCheckExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Tuple3<String, String, Long>> appStream = env
                .fromElements(
                        Tuple3.of("order-1", "app", 1000L),
                        Tuple3.of("order-2", "app", 2000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Tuple3<String, String, Long>>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple3<String, String, Long> in, long l) {
                                        return in.f2;
                                    }
                                })
                );

        SingleOutputStreamOperator<Tuple3<String, String, Long>> thirdPartyStream = env
                .fromElements(
                        Tuple3.of("order-1", "third-party", 3000L),
                        Tuple3.of("order-3", "third-party", 4000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Tuple3<String, String, Long>>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple3<String, String, Long> in, long l) {
                                        return in.f2;
                                    }
                                })
                );

        appStream.keyBy(r -> r.f0).connect(thirdPartyStream.keyBy(r -> r.f0)).process(new Match()).print();

        env.execute();
    }

    public static class Match extends CoProcessFunction<Tuple3<String, String, Long>, Tuple3<String, String, Long>, String> {
        // 声明状态变量用来保存app的支付事件
        private ValueState<Tuple3<String, String, Long>> appEvent;
        // 声明状态变量用来保存第三方平台的到账事件
        private ValueState<Tuple3<String, String, Long>> thirdPartyEvent;
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            appEvent = getRuntimeContext().getState(
                    new ValueStateDescriptor<Tuple3<String, String, Long>>("app", Types.TUPLE(
                            Types.STRING, Types.STRING, Types.LONG
                    ))
            );
            thirdPartyEvent = getRuntimeContext().getState(
                    new ValueStateDescriptor<Tuple3<String, String, Long>>("third-party", Types.TUPLE(
                            Types.STRING, Types.STRING, Types.LONG
                    ))
            );
        }

        @Override
        public void processElement1(Tuple3<String, String, Long> app, Context context, Collector<String> collector) throws Exception {
            if (thirdPartyEvent.value() != null) {
                collector.collect(app.f0 + " 对账成功");
                thirdPartyEvent.clear();
            } else {
                appEvent.update(app);
                context.timerService().registerEventTimeTimer(app.f2 + 5000L);
            }
        }

        @Override
        public void processElement2(Tuple3<String, String, Long> thirdParty, Context context, Collector<String> collector) throws Exception {
            if (appEvent.value() != null) {
                collector.collect(thirdParty.f0 + " 对账成功");
                appEvent.clear();
            } else {
                thirdPartyEvent.update(thirdParty);
                context.timerService().registerEventTimeTimer(thirdParty.f2 + 5000L);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            if (appEvent.value() != null) {
                out.collect(appEvent.value().f0 + "对账失败，订单的第三方支付信息未到");
                appEvent.clear();
            }
            if (thirdPartyEvent.value() != null) {
                out.collect(thirdPartyEvent.value().f0 + "对账失败，订单的app支付信息未到");
                thirdPartyEvent.clear();
            }
        }
    }
}

stream1.join(stream2)
    .where(<KeySelector>)
    .equalTo(<KeySelector>)
    .window(<WindowAssigner>)
    .apply(<JoinFunction>)

SELECT * FROM table1 t1, table2 t2 WHERE t1.id = t2.id;

// 基于窗口的join
public class WindowJoinExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<Tuple2<String, Long>> stream1 = env
                .fromElements(
                        Tuple2.of("a", 1000L),
                        Tuple2.of("b", 1000L),
                        Tuple2.of("a", 2000L),
                        Tuple2.of("b", 2000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Tuple2<String, Long>>forMonotonousTimestamps()
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                                            @Override
                                            public long extractTimestamp(Tuple2<String, Long> stringLongTuple2, long l) {
                                                return stringLongTuple2.f1;
                                            }
                                        }
                                )
                );

        DataStream<Tuple2<String, Long>> stream2 = env
                .fromElements(
                        Tuple2.of("a", 3000L),
                        Tuple2.of("b", 3000L),
                        Tuple2.of("a", 4000L),
                        Tuple2.of("b", 4000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Tuple2<String, Long>>forMonotonousTimestamps()
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                                            @Override
                                            public long extractTimestamp(Tuple2<String, Long> stringLongTuple2, long l) {
                                                return stringLongTuple2.f1;
                                            }
                                        }
                                )
                );

        stream1
                .join(stream2)
                .where(r -> r.f0)
                .equalTo(r -> r.f0)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .apply(new JoinFunction<Tuple2<String, Long>, Tuple2<String, Long>, String>() {
                    @Override
                    public String join(Tuple2<String, Long> left, Tuple2<String, Long> right) throws Exception {
                        return left + "=>" + right;
                    }
                })
                .print();

        env.execute();
    }
}

a.timestamp + lowerBound <= b.timestamp <= a.timestamp + upperBound

stream1
    .keyBy(<KeySelector>)
    .intervalJoin(stream2.keyBy(<KeySelector>))
    .between(Time.milliseconds(-2), Time.milliseconds(1))
    .process (new ProcessJoinFunction<Integer, Integer, String(){
        @Override
        public void processElement(Integer left, Integer right, Context ctx, Collector<String> out) {
            out.collect(left + "," + right);
        }
    });

// 基于间隔的join
public class IntervalJoinExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Tuple3<String, String, Long>> orderStream = env
                .fromElements(
                        Tuple3.of("Mary", "order", 20 * 60 * 1000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Tuple3<String, String, Long>>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple3<String, String, Long> e, long l) {
                                        return e.f2;
                                    }
                                })
                );

        SingleOutputStreamOperator<Tuple3<String, String, Long>> pvStream = env
                .fromElements(
                        Tuple3.of("Mary", "pv", 15 * 60 * 1000L),
                        Tuple3.of("Mary", "pv", 11 * 60 * 1000L),
                        Tuple3.of("Mary", "pv", 9 * 60 * 1000L),
                        Tuple3.of("Mary", "pv", 21 * 60 * 1000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Tuple3<String, String, Long>>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple3<String, String, Long> e, long l) {
                                        return e.f2;
                                    }
                                })
                );

        // a -> b
        // (lower, higher)
        orderStream
                .keyBy(r -> r.f0)
                .intervalJoin(pvStream.keyBy(r -> r.f0))
                .between(Time.minutes(-10), Time.minutes(5))
                .process(new ProcessJoinFunction<Tuple3<String, String, Long>, Tuple3<String, String, Long>, String>() {
                    @Override
                    public void processElement(Tuple3<String, String, Long> left, Tuple3<String, String, Long> right, Context context, Collector<String> collector) throws Exception {
                        collector.collect(left + "=>" + right);
                    }
                })
                .print();

        // b -> a
        // (-higher, -lower)
        pvStream
                .keyBy(r -> r.f0)
                .intervalJoin(orderStream.keyBy(r -> r.f0))
                .between(Time.minutes(-5), Time.minutes(10))
                .process(new ProcessJoinFunction<Tuple3<String, String, Long>, Tuple3<String, String, Long>, String>() {
                    @Override
                    public void processElement(Tuple3<String, String, Long> left, Tuple3<String, String, Long> right, Context context, Collector<String> collector) throws Exception {
                        collector.collect(right + "->" + left);
                    }
                })
                .print();

        env.execute();
    }
}

（1）算子任务接收到上游发来的数据；
（2）获取当前状态；
（3）根据业务逻辑进行计算，更新状态；
（4）得到计算结果，输出发送到下游任务。

1.状态的访问权限。我们知道Flink上的聚合和窗口操作，一般都是基于KeyedStream的，数据会按照key的哈希值进行分区，聚合处理的结果也应该是只对当前key有效。然而同一个分区（也就是slot）上执行的任务实例，可能会包含多个key的数据，它们同时访问和更改本地变量，就会导致计算结果错误。所以这时状态并不是单纯的本地变量。

2.容错性，也就是故障后的恢复。状态只保存在内存中显然是不够稳定的，我们需要将它持久化保存，做一个备份；在发生故障后可以从这个备份中恢复状态。

3.我们还应该考虑到分布式应用的横向扩展性。比如处理的数据量增大时，我们应该相应地对计算资源扩容，调大并行度。这时就涉及到了状态的重组调整。

  public interface ValueState<T> extends State {
T value() throws IOException;
void update(T value) throws IOException;
}

T value()：获取当前状态的值；
update(T value)：对状态进行更新，传入的参数value就是要覆写的状态值。

  public ValueStateDescriptor(String name, Class<T> typeClass) {
    super(name, typeClass, null);
}

Iterable<T> get()：获取当前的列表状态，返回的是一个可迭代类型Iterable<T>；
update(List<T> values)：传入一个列表values，直接对状态进行覆盖；
add(T value)：在状态列表中添加一个元素value；
addAll(List<T> values)：向列表中添加多个元素，以列表values形式传入。

UV get(UK key)：传入一个key作为参数，查询对应的value值；
put(UK key, UV value)：传入一个键值对，更新key对应的value值；
putAll(Map<UK, UV> map)：将传入的映射map中所有的键值对，全部添加到映射状态中；
remove(UK key)：将指定key对应的键值对删除；
boolean contains(UK key)：判断是否存在指定的key，返回一个boolean值。
另外，MapState也提供了获取整个映射相关信息的方法：
Iterable<Map.Entry<UK, UV>> entries()：获取映射状态中所有的键值对；
Iterable<UK> keys()：获取映射状态中所有的键（key），返回一个可迭代Iterable类型；
Iterable<UV> values()：获取映射状态中所有的值（value），返回一个可迭代Iterable类型；
boolean isEmpty()：判断映射是否为空，返回一个boolean值。

public ReducingStateDescriptor(
        String name, ReduceFunction<T> reduceFunction, Class<T> typeClass) {...}

ValueStateDescriptor<Long> descriptor = new ValueStateDescriptor<>(
"my state", // 状态名称
Types.LONG // 状态类型
);

ValueState<T> getState(ValueStateDescriptor<T>)
MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV>)
ListState<T> getListState(ListStateDescriptor<T>)
ReducingState<T> getReducingState(ReducingStateDescriptor<T>)
AggregatingState<IN, OUT> getAggregatingState(AggregatingStateDescriptor<IN, ACC, OUT>)

public static class MyFlatMapFunction extends RichFlatMapFunction<Long, String> {
    // 声明状态
private transient ValueState<Long> state;

    @Override
public void open(Configuration config) {
    // 在open生命周期方法中获取状态
        ValueStateDescriptor<Long> descriptor = new ValueStateDescriptor<>(
"my state", // 状态名称
Types.LONG // 状态类型 
);
        state = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void flatMap(Long input, Collector<String> out) throws Exception {
        // 访问状态
        Long currentState = state.value();
        currentState += 1;    // 状态数值加1
        // 更新状态
        state.update(currentState);
        if (currentState >= 100) {
            out.collect(“state: ” + currentState);
            state.clear();    // 清空状态
        }
    }
}

public class ValueStateExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                )
                .keyBy(r -> r.user)
                .process(new KeyedProcessFunction<String, Event, String>() {
                    private ValueState<Long> valueState;
                    private ValueState<Long> timerTs;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        valueState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("pv", Types.LONG));
                    }

                    @Override
                    public void processElement(Event value, Context ctx, Collector<String> out) throws Exception {
                        if (valueState.value() == null) {
                            valueState.update(1L);
                        } else {
                            valueState.update(valueState.value() + 1L);
                        }
                        if (timerTs.value() == null) {
ctx.timerService().registerEventTimeTimer(value.timestamp + 10 * 1000L);
                            timerTs.update(value.timestamp + 10 * 1000L);
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        super.onTimer(timestamp, ctx, out);
                        out.collect("用户 " + ctx.getCurrentKey() + " 的PV是：" + valueState.value());
                        timerTs.clear();
                    }
                })
                .print();

        env.execute();
    }
}

// 使用KeyedProcessFunction模拟滚动窗口
public class FakeWindowExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                    @Override
                    public long extractTimestamp(Event event, long l) {
                        return event.timestamp;
                    }
                }))
                .keyBy(r -> r.url)
                .process(new FakeWindow(5000L))
                .print();

        env.execute();
    }

    public static class FakeWindow extends KeyedProcessFunction<String, Event, String> {
        // 窗口的开始时间 -> 窗口中的pv
        private MapState<Long, Long> mapState;
        // 滚动窗口的长度
        private Long windowSize;

        public FakeWindow(Long windowSize) {
            this.windowSize = windowSize;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            mapState = getRuntimeContext().getMapState(
                    new MapStateDescriptor<Long, Long>("window-pv", Types.LONG, Types.LONG)
            );
        }

        @Override
        public void processElement(Event event, Context context, Collector<String> collector) throws Exception {
            long windowStart = event.timestamp - event.timestamp % windowSize;
            long windowEnd = windowStart + windowSize;
            context.timerService().registerEventTimeTimer(windowEnd – 1);
            if (mapState.contains(windowStart)) {
                long pv = mapState.get(windowStart);
                mapState.put(windowStart, pv + 1L);
            } else {
                mapState.put(windowStart, 1L);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            long start = timestamp + 1L - windowSize;
            long end = timestamp + 1L;
            out.collect(ctx.getCurrentKey() + ":" + new Timestamp(start) + "~" + new Timestamp(end) + ":" + mapState.get(start));
// 删除窗口，因为窗口的默认操作是计算完成以后销毁窗口
            mapState.remove(start);
        }
    }

public class AggregateStateExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env
                .addSource(new SourceFunction<Tuple2<String, Integer>>() {
                    private boolean running = true;
                    private Random random = new Random();
                    @Override
                    public void run(SourceContext<Tuple2<String, Integer>> sourceContext) throws Exception {
                        while (true) {
                            sourceContext.collect(Tuple2.of("key", random.nextInt()));
                        }
                    }

                    @Override
                    public void cancel() {
                        running = false;
                    }
                })
                .keyBy(r -> r.f0)
                .flatMap(new CountFunction())
                .print();
        env.execute();
    }

    public static class CountFunction extends RichFlatMapFunction<Tuple2<String, Integer>, Integer> {
        private int count = 0;
        // 声明聚合状态变量
        private AggregatingState<Tuple2<String, Integer>, Integer> aggregatingState;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            AggregatingStateDescriptor<Tuple2<String, Integer>, Integer, Integer> descriptor = new AggregatingStateDescriptor<Tuple2<String, Integer>, Integer, Integer>(
                    "aggregatingState", new AggregateFunction<Tuple2<String, Integer>, Integer, Integer>() {
                @Override
                public Integer createAccumulator() {
                    return 0;
                }

                @Override
                public Integer add(Tuple2<String, Integer> value, Integer accumulator) {
                    return accumulator + 1;
                }

                @Override
                public Integer getResult(Integer accumulator) {
                    return accumulator;
                }

                @Override
                public Integer merge(Integer a, Integer b) {
                    return a + b;
                }
            }, Types.INT);
            aggregatingState = getRuntimeContext().getAggregatingState(descriptor);
        }

        @Override
        public void flatMap(Tuple2<String, Integer> value, Collector<Integer> out) throws Exception {
            count++;
            if (count % 1000 == 0) {
                out.collect(aggregatingState.get());
                aggregatingState.clear(); // 清空状态变量
            } else {
                // 增量更新AggregatingState，这里每来一个新元素，对ACC累加1
                aggregatingState.add(value);
            }
        }
    }
}

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.seconds(10))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .build();
ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("my state", String.class);
stateDescriptor.enableTimeToLive(ttlConfig);

.newBuilder()
状态TTL配置的构造器方法，必须调用，返回一个Builder之后再调用.build()方法就可以得到StateTtlConfig了。方法需要传入一个Time作为参数，这就是设定的状态生存时间。
.setUpdateType()
设置更新类型。更新类型指定了什么时候更新状态失效时间，这里的OnCreateAndWrite表示只有创建状态和更改状态（写操作）时更新失效时间。另一种类型OnReadAndWrite则表示无论读写操作都会更新失效时间，也就是只要对状态进行了访问，就表明它是活跃的，从而延长生存时间。这个配置默认为OnCreateAndWrite。
.setStateVisibility()
设置状态的可见性。所谓的“状态可见性”，是指因为清除操作并不是实时的，所以当状态过期之后还有可能继续存在，这时如果对它进行访问，能否正常读取到就是一个问题了。这里设置的NeverReturnExpired是默认行为，表示从不返回过期值，也就是只要过期就认为它已经被清除了，应用不能继续读取；这在处理会话或者隐私数据时比较重要。对应的另一种配置是ReturnExpireDefNotCleanedUp，就是如果过期状态还存在，就返回它的值。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getEnvironment();
env.enableCheckpointing(1000);

# 默认状态后端
state.backend: hashmap
# 存放检查点的文件路径
state.checkpoints.dir: hdfs://namenode:40010/flink/checkpoints
这里的state.checkpoints.dir配置项，定义了状态后端将检查点和元数据写入的目录。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new HashMapStateBackend());

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new EmbeddedRocksDBStateBackend());

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>

“hello”，“world”，“hello”，“flink”，“hello”，“world”，“hello”，“flink”…

SingleOutputStreamOperator<Tuple2<String, Long>> wordCountStream = 
env.addSource(...)
        .map(word -> Tuple2.of(word, 1L))
        .returns(Types.TUPLE(Types.STRING, Types.LONG));
        .keyBy(t -> t.f0);
        .sum(1);

StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
// 每隔1秒启动一次检查点保存
env.enableCheckpointing(1000);

// 配置存储检查点到JobManager堆内存
env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());
// 配置存储检查点到文件系统
env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("hdfs://namenode:40010/flink/checkpoints"));

1.检查点模式（CheckpointingMode）
设置检查点一致性的保证级别，有“精确一次”（exactly-once）和“至少一次”（at-least-once）两个选项。默认级别为exactly-once，而对于大多数低延迟的流处理程序，at-least-once就够用了，而且处理效率会更高。关于一致性级别，我们会在10.2节继续展开。
2.超时时间（checkpointTimeout）
用于指定检查点保存的超时时间，超时没完成就会被丢弃掉。传入一个长整型毫秒数作为参数，表示超时时间。
3.最小间隔时间（minPauseBetweenCheckpoints）
用于指定在上一个检查点完成之后，检查点协调器（checkpoint coordinator）最快等多久可以出发保存下一个检查点的指令。这就意味着即使已经达到了周期触发的时间点，只要距离上一个检查点完成的间隔不够，就依然不能开启下一次检查点的保存。这就为正常处理数据留下了充足的间隙。当指定这个参数时，maxConcurrentCheckpoints的值强制为1。
4.最大并发检查点数量（maxConcurrentCheckpoints）
用于指定运行中的检查点最多可以有多少个。由于每个任务的处理进度不同，完全可能出现后面的任务还没完成前一个检查点的保存、前面任务已经开始保存下一个检查点了。这个参数就是限制同时进行的最大数量。
如果前面设置了minPauseBetweenCheckpoints，则maxConcurrentCheckpoints这个参数就不起作用了。
5.开启外部持久化存储（enableExternalizedCheckpoints）
用于开启检查点的外部持久化，而且默认在作业失败的时候不会自动清理，如果想释放空间需要自己手工清理。里面传入的参数ExternalizedCheckpointCleanup指定了当作业取消的时候外部的检查点该如何清理。
DELETE_ON_CANCELLATION：在作业取消的时候会自动删除外部检查点，但是如果是作业失败退出，则会保留检查点。
RETAIN_ON_CANCELLATION：作业取消的时候也会保留外部检查点。
6.检查点异常时是否让整个任务失败（failOnCheckpointingErrors）
用于指定在检查点发生异常的时候，是否应该让任务直接失败退出。默认为true，如果设置为false，则任务会丢弃掉检查点然后继续运行。
7..不对齐检查点（enableUnalignedCheckpoints）
不再执行检查点的分界线对齐操作，启用之后可以大大减少产生背压时的检查点保存时间。这个设置要求检查点模式（CheckpointingMode）必须为exctly-once，并且并发的检查点个数为1。

StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();

// 启用检查点，间隔时间1秒
env.enableCheckpointing(1000);
CheckpointConfig checkpointConfig = env.getCheckpointConfig();
// 设置精确一次模式
checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 最小间隔时间500毫秒
checkpointConfig.setMinPauseBetweenCheckpoints(500);
// 超时时间1分钟
checkpointConfig.setCheckpointTimeout(60000);
// 同时只能有一个检查点
checkpointConfig.setMaxConcurrentCheckpoints(1);
// 开启检查点的外部持久化保存，作业取消后依然保留
checkpointConfig.enableExternalizedCheckpoints(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
// 启用不对齐的检查点保存方式
checkpointConfig.enableUnalignedCheckpoints();
// 设置检查点存储，可以直接传入一个String，指定文件系统的路径
checkpointConfig.setCheckpointStorage("hdfs://my/checkpoint/dir")

1.版本管理和归档存储
对重要的节点进行手动备份，设置为某一版本，归档（archive）存储应用程序的状态。
2.更新Flink版本
目前Flink的底层架构已经非常稳定，所以当Flink版本升级时，程序本身一般是兼容的。这时不需要重新执行所有的计算，只要创建一个保存点，停掉应用、升级Flink后，从保存点重启就可以继续处理了。
3.更新应用程序
我们不仅可以在应用程序不变的时候，更新Flink版本；还可以直接更新应用程序。前提是程序必须是兼容的，也就是说更改之后的程序，状态的拓扑结构和数据类型都是不变的，这样才能正常从之前的保存点去加载。
这个功能非常有用。我们可以及时修复应用程序中的逻辑bug，更新之后接着继续处理；也可以用于有不同业务逻辑的场景，比如A/B测试等等。
4.调整并行度
如果应用运行的过程中，发现需要的资源不足、或者已经有了大量剩余，也可以通过从保存点重启的方式，将应用程序的并行度增大或减小。
5.暂停应用程序
有时候我们不需要调整集群或者更新程序，只是单纯地希望把应用暂停、释放一些资源来处理更重要的应用程序。使用保存点就可以灵活实现应用的暂停和重启，可以对有限的集群资源做最好的优化配置。
需要注意的是，保存点能够在程序更改的时候依然兼容，前提是状态的拓扑结构和数据类型不变。我们知道保存点中状态都是以算子ID-状态名称这样的key-value组织起来的，算子ID可以在代码中直接调用SingleOutputStreamOperator的.uid()方法来进行指定：

DataStream<String> stream = env
  .addSource(new StatefulSource())
  .uid("source-id")
  .map(new StatefulMapper())
  .uid("mapper-id")
  .print();

bin/flink savepoint :jobId [:targetDirectory]

state.savepoints.dir: hdfs:///flink/savepoints

env.setDefaultSavepointDir("hdfs:///flink/savepoints");

bin/flink stop --savepointPath [:targetDirectory] :jobId

bin/flink run -s :savepointPath [:runArgs]

1.最多一次（AT-MOST-ONCE） 
当任务发生故障时，最简单的做法就是直接重启，别的什么都不干；既不恢复丢失的状态，也不重放丢失的数据。每个数据在正常情况下会被处理一次，遇到故障时就会丢掉，所以就是“最多处理一次”。
我们发现，如果数据可以直接被丢掉，那其实就是没有任何操作来保证结果的准确性；所以这种类型的保证也叫“没有保证”。尽管看起来比较糟糕，不过如果我们的主要诉求是“快”，而对近似正确的结果也能接受，那这也不失为一种很好的解决方案。
2.至少一次（AT-LEAST-ONCE） 
在实际应用中，我们一般会希望至少不要丢掉数据。这种一致性级别就叫做“至少一次”（at-least-once），就是说是所有数据都不会丢，肯定被处理了；不过不能保证只处理一次，有些数据会被重复处理。
在有些场景下，重复处理数据是不影响结果的正确性的，这种操作具有“幂等性”。比如，如果我们统计电商网站的UV，需要对每个用户的访问数据进行去重处理，所以即使同一个数据被处理多次，也不会影响最终的结果，这时使用at-least-once语义是完全没问题的。当然，如果重复数据对结果有影响，比如统计的是PV，或者之前的统计词频word count，使用at-least-once语义就可能会导致结果的不一致了。
为了保证达到at-least-once 的状态一致性，我们需要在发生故障时能够重放数据。最常见的做法是，可以用持久化的事件日志系统，把所有的事件写入到持久化存储中。这时只要记录一个偏移量，当任务发生故障重启后，重置偏移量就可以重放检查点之后的数据了。Kafka就是这种架构的一个典型实现。
3.精确一次（EXACTLY-ONCE） 
最严格的一致性保证，就是所谓的“精确一次”（exactly-once，有时也译作“恰好一次”）。这也是最难实现的状态一致性语义。exactly-once意味着所有数据不仅不会丢失，而且只被处理一次，不会重复处理。也就是说对于每一个数据，最终体现在状态和输出结果上，只能有一次统计。
exactly-once可以真正意义上保证结果的绝对正确，在发生故障恢复后，就好像从未发生过故障一样。

幂等写入
事务写入

（1）预写日志（write-ahead-log，WAL）
我们发现，事务提交是需要外部存储系统支持事务的，否则没有办法真正实现写入的回撤。那对于一般不支持事务的存储系统，能够实现事务写入呢？
预写日志（WAL）就是一种非常简单的方式。具体步骤是：
①先把结果数据作为日志（log）状态保存起来
②进行检查点保存时，也会将这些结果数据一并做持久化存储
③在收到检查点完成的通知时，将所有结果一次性写入外部系统。
我们会发现，这种方式类似于检查点完成时做一个批处理，一次性的写入会带来一些性能上的问题；而优点就是比较简单，由于数据提前在状态后端中做了缓存，所以无论什么外部存储系统，理论上都能用这种方式一批搞定。在Flink中DataStream API提供了一个模板类GenericWriteAheadSink，用来实现这种事务型的写入方式。
需要注意的是，预写日志这种一批写入的方式，有可能会写入失败；所以在执行写入动作之后，必须等待发送成功的返回确认消息。在成功写入所有数据后，在内部再次确认相应的检查点，这才代表着检查点的真正完成。这里需要将确认信息也进行持久化保存，在故障恢复时，只有存在对应的确认信息，才能保证这批数据已经写入，可以恢复到对应的检查点位置。
但这种“再次确认”的方式，也会有一些缺陷。如果我们的检查点已经成功保存、数据也成功地一批写入到了外部系统，但是最终保存确认信息时出现了故障，Flink最终还是会认为没有成功写入。于是发生故障时，不会使用这个检查点，而是需要回退到上一个；这样就会导致这批数据的重复写入。

（2）两阶段提交（two-phase-commit，2PC）
前面提到的各种实现exactly-once的方式，多少都有点缺陷；而更好的方法就是传说中的两阶段提交（2PC）。
顾名思义，它的想法是分成两个阶段：先做“预提交”，等检查点完成之后再正式提交。这种提交方式是真正基于事务的，它需要外部系统提供事务支持。
具体的实现步骤为：
①当第一条数据到来时，或者收到检查点的分界线时，Sink任务都会启动一个事务。
②接下来接收到的所有数据，都通过这个事务写入外部系统；这时由于事务没有提交，所以数据尽管写入了外部系统，但是不可用，是“预提交”的状态。
③当Sink任务收到JobManager发来检查点完成的通知时，正式提交事务，写入的结果就真正可用了。
当中间发生故障时，当前未提交的事务就会回滚，于是所有写入外部系统的数据也就实现了撤回。这种两阶段提交（2PC）的方式充分利用了Flink现有的检查点机制：分界线的到来，就标志着开始一个新事务；而收到来自JobManager的checkpoint成功的消息，就是提交事务的指令。每个结果数据的写入，依然是流式的，不再有预写日志时批处理的性能问题；最终提交时，也只需要额外发送一个确认信息。所以2PC协议不仅真正意义上实现了exactly-once，而且通过搭载Flink的检查点机制来实现事务，只给系统增加了很少的开销。

1.外部系统必须提供事务支持，或者Sink任务必须能够模拟外部系统上的事务。
2.在检查点的间隔期间里，必须能够开启一个事务并接受数据写入。
3.在收到检查点完成的通知之前，事务必须是“等待提交”的状态。在故障恢复的情况下，这可能需要一些时间。如果这个时候外部系统关闭事务（例如超时了），那么未提交的数据就会丢失。
4.Sink任务必须能够在进程失败后恢复事务。
5.提交事务必须是幂等操作。也就是说，事务的重复提交应该是无效的。

（1）Flink内部
Flink内部可以通过检查点机制保证状态和处理结果的exactly-once语义。
（2）输入端
输入数据源端的Kafka可以对数据进行持久化保存，并可以重置偏移量（offset）。所以我们可以在Source任务（FlinkKafkaConsumer）中将当前读取的偏移量保存为算子状态，写入到检查点中；当发生故障时，从检查点中读取恢复状态，并由连接器FlinkKafkaConsumer向Kafka重新提交偏移量，就可以重新消费数据、保证结果的一致性了。
（3）输出端
输出端保证exactly-once的最佳实现，当然就是两阶段提交（2PC）。作为与Flink天生一对的Kafka，自然需要用最强有力的一致性保证来证明自己。

public class FlinkKafkaProducer<IN> extends TwoPhaseCommitSinkFunction<IN, FlinkKafkaProducer.KafkaTransactionState, FlinkKafkaProducer.KafkaTransactionContext> {
...
}

（1）必须启用检查点
（2）在FlinkKafkaProducer的构造函数中传入参数Semantic.EXACTLY_ONCE
（3）配置Kafka读取数据的消费者的隔离级别
这里所说的Kafka，是写入的外部系统。预提交阶段数据已经写入，只是被标记为“未提交”（uncommitted），而Kafka中默认的隔离级别isolation.level是read_uncommitted，也就是可以读取未提交的数据。这样一来，外部应用就可以直接消费未提交的数据，对于事务性的保证就失效了。所以应该将隔离级别配置
为read_committed，表示消费者遇到未提交的消息时，会停止从分区中消费数据，直到消息被标记为已提交才会再次恢复消费。当然，这样做的话，外部应用消费数据就会有显著的延迟。
（4）事务超时配置
Flink的Kafka连接器中配置的事务超时时间transaction.timeout.ms默认是1小时，而Kafka集群配置的事务最大超时时间transaction.max.timeout.ms默认是15分钟。所以在检查点保存时间很长时，有可能出现Kafka已经认为事务超时了，丢弃了预提交的数据；而Sink任务认为还可以继续等待。如果接下来检查点保存成功，发生故障后回滚到这个检查点的状态，这部分数据就被真正丢掉了。所以这两个超时时间，前者应该小于等于后者。