flink1.10.1--java 版--尚硅谷-第六章 Flink Window+wartermark+状态+容错

淡淡的烟草味﹌ 2023-01-20 09:42 9阅读 0赞

# 第六章 Flink 中的 Window #

## 6.1 Window ##

### 6.1.1 Window 概述 ###

![在这里插入图片描述][20210512104315351.png]  
桶的概念，不是批处理

### 6.1.2 Window 类型 ###

![在这里插入图片描述][20210512104514621.png]  
**时间窗口的设置是左闭右开\[ ），及包含开时间进入窗口的数据不包含结束时间进入窗口的数据**  
![在这里插入图片描述][20210512104622629.png]

##### 1. 滚动窗口（Tumbling Windows） #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70]

#### 2. 滑动窗口（Sliding Windows） ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 2]

#### 3. 会话窗口（Session Windows） ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 3]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 4]

## 6.2 Window API ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 5]

### 6.2.1 TimeWindow ###

![在这里插入图片描述][20210512104909438.png]

DataStream<Integer> resultStream = dataStream.keyBy("id")
 //滑动窗口的简写方式
 .timeWindow(Time.seconds(15),Time.seconds(10))
 //滚动窗口的简写方式
 // .timeWindow(Time.seconds(15))

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 6]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 7]

#### 1. 滚动窗口 ####

![在这里插入图片描述][20210512105148796.png]

DataStream<Tuple2<String, Double>> minTempPerWindowStream = dataStream
 .map(new MapFunction<SensorReading, Tuple2<String, Double>>() { 
 @Override
 public Tuple2<String, Double> map(SensorReading value) throws
 Exception { 
 return new Tuple2<>(value.getId(), value.getTemperature());
 }
 })
 .keyBy(data -> data.f0)
 .timeWindow(Time.seconds(15))
 .minBy(1);

时间间隔可以通过 Time.milliseconds(x)，Time.seconds(x)，Time.minutes(x)等其  
中的一个来指定。

#### 2. 滑动窗口（SlidingEventTimeWindows） ####

![在这里插入图片描述][20210512105315706.png]

DataStream<SensorReading> minTempPerWindowStream = dataStream
 .keyBy(SensorReading::getId)
 .timeWindow( Time.seconds(15), Time.seconds(5) )
 .minBy("temperature");

时间间隔可以通过 Time.milliseconds(x)，Time.seconds(x)，Time.minutes(x)等其  
中的一个来指定。

### 6.2.2 CountWindow ###

![在这里插入图片描述][20210512105452926.png]

#### 1 滚动窗口 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 8]

#### 2 滑动窗口 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 9]

### 6.2.3 window function ###

之前讲的其他聚合转换入sum,reduce都是有状态的属于增量聚合。  
增量聚合是来一个计算一个，对窗口中的计算结果有状态，只是不输出，实时性相对于全窗口函数跟高。  
全量聚合函数，更灵活些可以获取更多信息，可以自定义输出的类型  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 10]

### 6.2.4 其它可选 API ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 11]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 12]

package com.atguigu.apitest.window;/** * Copyright (c) 2018-2028 尚硅谷 All Rights Reserved * * Project: FlinkTutorial * Package: com.atguigu.apitest.window * Version: 1.0 * * Created by wushengran on 2020/11/9 14:37 */
 
 import com.atguigu.apitest.beans.SensorReading;
 import org.apache.commons.collections.IteratorUtils;
 import org.apache.flink.api.common.functions.AggregateFunction;
 import org.apache.flink.api.common.functions.ReduceFunction;
 import org.apache.flink.api.java.tuple.Tuple;
 import org.apache.flink.api.java.tuple.Tuple3;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
 import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
 import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
 import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
 import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
 import org.apache.flink.streaming.api.windowing.time.Time;
 import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
 import org.apache.flink.util.Collector;
 import org.apache.flink.util.OutputTag;
 
 import java.util.Arrays;
 import java.util.Collections;
 
 /** * @ClassName: WindowTest1_TimeWindow * @Description: * @Author: wushengran on 2020/11/9 14:37 * @Version: 1.0 */
 public class WindowTest1_TimeWindow { 
 public static void main(String[] args) throws Exception { 
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);
 
 // // 从文件读取数据
 // DataStream<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
 
 // socket文本流
 DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
 
 // 转换成SensorReading类型
 DataStream<SensorReading> dataStream = inputStream.map(line -> { 
 String[] fields = line.split(",");
 return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
 });
 
 // 开窗测试
 
 // 1. 增量聚合函数
 DataStream<Integer> resultStream = dataStream.keyBy("id")
 // .countWindow(10, 2);
 // .window(EventTimeSessionWindows.withGap(Time.minutes(1)));
 // .window(TumblingProcessingTimeWindows.of(Time.seconds(15)))
 //滑动窗口的简写方式
 //.timeWindow(Time.seconds(15),Time.seconds(10))
 //滚动窗口的简写方式
 .timeWindow(Time.seconds(15))
 .aggregate(new AggregateFunction<SensorReading, Integer, Integer>() { 
 @Override
 public Integer createAccumulator() { 
 return 0;
 }
 
 @Override
 public Integer add(SensorReading 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;
 }
 });
 
 // 2. 全窗口函数
 SingleOutputStreamOperator<Tuple3<String, Long, Integer>> resultStream2 = dataStream.keyBy("id")
 .timeWindow(Time.seconds(15))
 // .process(new ProcessWindowFunction<SensorReading, Object, Tuple, TimeWindow>() { 
 // })
 .apply(new WindowFunction<SensorReading, Tuple3<String, Long, Integer>, Tuple, TimeWindow>() { 
 @Override
 public void apply(Tuple tuple, TimeWindow window, Iterable<SensorReading> input, Collector<Tuple3<String, Long, Integer>> out) throws Exception { 
 String id = tuple.getField(0);
 Long windowEnd = window.getEnd();
 Integer count = IteratorUtils.toList(input.iterator()).size();
 out.collect(new Tuple3<>(id, windowEnd, count));
 }
 });
 
 // 3. 其它可选API
 OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>("late") { 
 };
 
 SingleOutputStreamOperator<SensorReading> sumStream = dataStream.keyBy("id")
 .timeWindow(Time.seconds(15))
 // .trigger()
 // .evictor()
 .allowedLateness(Time.minutes(1))
 .sideOutputLateData(outputTag)
 .sum("temperature");
 
 sumStream.getSideOutput(outputTag).print("late");
 
 resultStream2.print();
 
 
 env.execute();
 
 
 }
 }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 13]

# 第七章 时间语义与 Wartermark #

## 7.1 Flink 中的时间语义 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 14]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 15]

## 7.2 EventTime 的引入 ##

![在这里插入图片描述][20210512143842849.png]

StreamExecutionEnvironment env = 
    StreamExecutionEnvironment.getExecutionEnvironment
    // 从调用时刻开始给 env 创建的每一个 stream 追加时间特征
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

## 7.3 Watermark(水位线) ##

### 7.3.1 基本概念 ###

![在这里插入图片描述][20210512143944573.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 16]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 17]  
![在这里插入图片描述][20210512144408989.png]  
**watermark 的特点**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 18]

**watermark 的传递**  
watermark往下游是以广播的方式传递。  
以最小的watermark为准。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 19]

### 在代码中设置 Event Time ###

![在这里插入图片描述][2021051214584323.png]  
执行环境指定为事件时间语义，及窗口的关闭会根据事件时间关闭  
![在这里插入图片描述][20210512145801847.png]

### 7.3.2 Watermark 的引入 ###

watermark 的引入很简单，对于乱序数据，最常见的引用方式如下

前提是要指定执行环境指定为事件时间语义  
![在这里插入图片描述][20210512145801847.png]  
代码中watermark 的引入，指定每条消息的时间时间  
![在这里插入图片描述][20210512150500240.png]  
![在这里插入图片描述][20210512144720840.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 20]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 21]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 22]

# 第九章 Flink 状态管理和容错机制 #

## 9.2 状态一致性 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 23]

### 9.2.1 一致性级别 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 24]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 25]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 26]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 27]  
**幂等写入（Idempotent Writes）：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 28]  
**事务写入（Transactional Writes）：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 29]  
**预写日志（Write-Ahead-Log，WAL）**

先把数据当成日志先保存，等checkpoint完成后再一次性写入sink系统，缺点是需要等待相当于一批一批的写入了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 30]  
**两阶段提交（Two-Phase-Commit，2PC）**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 31]  
**2PC 对外部 sink 系统的要求**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 32]

### 9.2.2 端到端（end-to-end）状态一致性 ###

## Flink 状态管理 ##

![在这里插入图片描述][20210512155103110.png]

Flink 中的状态：  
• 算子状态（Operatior State）  
• 键控状态（Keyed State）针对每一个key有一个状态  
• 状态后端（State Backends）

**要想使用状态，必须先向flink注册状态，来告诉flink我有一个状态需要你帮忙保存。**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 33]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 34]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 35]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 36]

### 9.3.1 算子状态（operator state） ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 37]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 38]

package com.atguigu.apitest.state;/** * Copyright (c) 2018-2028 尚硅谷 All Rights Reserved * * Project: FlinkTutorial * Package: com.atguigu.apitest.state * Version: 1.0 * * Created by wushengran on 2020/11/10 15:30 */
 
 import com.atguigu.apitest.beans.SensorReading;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
 import java.util.Collections;
 import java.util.List;
 
 /** * @ClassName: StateTest1_OperatorState * @Description: * @Author: wushengran on 2020/11/10 15:30 * @Version: 1.0 */
 public class StateTest1_OperatorState { 
 public static void main(String[] args) throws Exception{ 
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);
 
 // socket文本流
 DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
 
 // 转换成SensorReading类型
 DataStream<SensorReading> dataStream = inputStream.map(line -> { 
 String[] fields = line.split(",");
 return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
 });
 
 // 定义一个有状态的map操作，统计当前分区数据个数
 SingleOutputStreamOperator<Integer> resultStream = dataStream.map(new MyCountMapper());
 
 resultStream.print();
 
 env.execute();
 }
 
 // 自定义MapFunction
 //实现ListCheckpointed接口来保存和重启读取状态值
 public static class MyCountMapper implements MapFunction<SensorReading, Integer>, ListCheckpointed<Integer>{ 
 // 定义一个本地变量，作为算子状态，只是一个本地内存值，如果不保存到状态中故障重启后状态值就不存在了
 private Integer count = 0;
 
 @Override
 public Integer map(SensorReading value) throws Exception { 
 count++;
 return count;
 }
 
 //保存状态值
 @Override
 public List<Integer> snapshotState(long checkpointId, long timestamp) throws Exception { 
 return Collections.singletonList(count);
 }
 
 //发生故障的启后从状态读取之前保存的值
 @Override
 public void restoreState(List<Integer> state) throws Exception { 
 for( Integer num: state )
 count += num;
 }
 }
 }

### 9.3.2 键控状态（keyed state） ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 39]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 40]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 41]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 42]

package com.atguigu.apitest.state;/** * Copyright (c) 2018-2028 尚硅谷 All Rights Reserved * * Project: FlinkTutorial * Package: com.atguigu.apitest.state * Version: 1.0 * * Created by wushengran on 2020/11/10 15:49 */
 
 import com.atguigu.apitest.beans.SensorReading;
 import org.apache.flink.api.common.functions.RichMapFunction;
 import org.apache.flink.api.common.state.*;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
 /** * @ClassName: StateTest2_KeyedState * @Description: * @Author: wushengran on 2020/11/10 15:49 * @Version: 1.0 */
 public class StateTest2_KeyedState { 
 public static void main(String[] args) throws Exception{ 
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);
 
 // socket文本流
 DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
 
 // 转换成SensorReading类型
 DataStream<SensorReading> dataStream = inputStream.map(line -> { 
 String[] fields = line.split(",");
 return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
 });
 
 // 定义一个有状态的map操作，统计当前sensor数据个数
 SingleOutputStreamOperator<Integer> resultStream = dataStream
 .keyBy("id")
 .map( new MyKeyCountMapper() );
 
 resultStream.print();
 
 env.execute();
 }
 
 // 自定义RichMapFunction
 public static class MyKeyCountMapper extends RichMapFunction<SensorReading, Integer>{ 
 private ValueState<Integer> keyCountState;
 
 // 其它类型状态的声明
 private ListState<String> myListState;
 private MapState<String, Double> myMapState;
 private ReducingState<SensorReading> myReducingState;
 
 @Override
 public void open(Configuration parameters) throws Exception { 
 keyCountState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("key-count", Integer.class, 0));
 
 myListState = getRuntimeContext().getListState(new ListStateDescriptor<String>("my-list", String.class));
 myMapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Double>("my-map", String.class, Double.class));
 // myReducingState = getRuntimeContext().getReducingState(new ReducingStateDescriptor<SensorReading>())
 }
 
 @Override
 public Integer map(SensorReading value) throws Exception { 
 // 其它状态API调用
 // list state
 for(String str: myListState.get()){ 
 System.out.println(str);
 }
 myListState.add("hello");
 
 // map state
 myMapState.get("1");
 myMapState.put("2", 12.3);
 myMapState.remove("2");
 //清空状态
 myMapState.clear();
 
 // reducing state
 // myReducingState.add(value);
 
 
 //Value State
 Integer count = keyCountState.value();
 count++;
 keyCountState.update(count);
 return count;
 }
 }
 }

### 检测温度跳变输出报警实例 ###

我们可以利用 Keyed State，实现这样一个需求：检测传感器的温度值，如果连续的两个温度差值超过 10 度，就输出报警。

package com.atguigu.apitest.state;/** * Copyright (c) 2018-2028 尚硅谷 All Rights Reserved * * Project: FlinkTutorial * Package: com.atguigu.apitest.state * Version: 1.0 * * Created by wushengran on 2020/11/10 16:33 */
 
 import com.atguigu.apitest.beans.SensorReading;
 import org.apache.flink.api.common.functions.FlatMapFunction;
 import org.apache.flink.api.common.functions.RichFlatMapFunction;
 import org.apache.flink.api.common.functions.RichMapFunction;
 import org.apache.flink.api.common.state.ValueState;
 import org.apache.flink.api.common.state.ValueStateDescriptor;
 import org.apache.flink.api.java.tuple.Tuple3;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.util.Collector;
 
 /** * @ClassName: StateTest3_KeyedStateApplicationCase * @Description: * @Author: wushengran on 2020/11/10 16:33 * @Version: 1.0 */
 public class StateTest3_KeyedStateApplicationCase { 
 public static void main(String[] args) throws Exception{ 
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);
 
 // socket文本流
 DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
 
 // 转换成SensorReading类型
 DataStream<SensorReading> dataStream = inputStream.map(line -> { 
 String[] fields = line.split(",");
 return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
 });
 
 // 定义一个flatmap操作，检测温度跳变，输出报警
 SingleOutputStreamOperator<Tuple3<String, Double, Double>> resultStream = dataStream.keyBy("id")
 .flatMap(new TempChangeWarning(10.0));
 
 resultStream.print();
 
 env.execute();
 }
 
 // 实现自定义函数类
 public static class TempChangeWarning extends RichFlatMapFunction<SensorReading, Tuple3<String, Double, Double>>{ 
 // 私有属性，温度跳变阈值
 private Double threshold;
 
 public TempChangeWarning(Double threshold) { 
 this.threshold = threshold;
 }
 
 // 定义状态，保存上一次的温度值
 private ValueState<Double> lastTempState;
 
 @Override
 public void open(Configuration parameters) throws Exception { 
 lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<Double>("last-temp", Double.class));
 }
 
 @Override
 public void flatMap(SensorReading value, Collector<Tuple3<String, Double, Double>> out) throws Exception { 
 // 获取状态
 Double lastTemp = lastTempState.value();
 
 // 如果状态不为null，那么就判断两次温度差值
 if( lastTemp != null ){ 
 Double diff = Math.abs( value.getTemperature() - lastTemp );
 if( diff >= threshold )
 out.collect(new Tuple3<>(value.getId(), lastTemp, value.getTemperature()));
 }
 
 // 更新状态
 lastTempState.update(value.getTemperature());
 }
 
 @Override
 public void close() throws Exception { 
 lastTempState.clear();
 }
 }
 }

### 9.3.3状态后端（State Backends） ###

保存状态  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 43]  
**状态后端保存方式(state backend)：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 44]  
设置状态后端为 FsStateBackend，并配置检查点和重启策略：

package com.atguigu.apitest.state;/** * Copyright (c) 2018-2028 尚硅谷 All Rights Reserved * * Project: FlinkTutorial * Package: com.atguigu.apitest.state * Version: 1.0 * * Created by wushengran on 2020/11/11 9:30 */
 
 import com.atguigu.apitest.beans.SensorReading;
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
 import org.apache.flink.runtime.state.filesystem.FsStateBackend;
 import org.apache.flink.runtime.state.memory.MemoryStateBackend;
 import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
 /** * @ClassName: StateTest4_FaultTolerance * @Description: * @Author: wushengran on 2020/11/11 9:30 * @Version: 1.0 */
 public class StateTest4_FaultTolerance { 
 public static void main(String[] args) throws Exception{ 
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);
 
 // 1. 状态后端配置
 env.setStateBackend( new MemoryStateBackend());
 env.setStateBackend( new FsStateBackend(""));
 env.setStateBackend( new RocksDBStateBackend(""));
 
 // 2. 检查点配置
 env.enableCheckpointing(300);
 
 // 高级选项
 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
 env.getCheckpointConfig().setCheckpointTimeout(60000L);
 env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(100L);
 env.getCheckpointConfig().setPreferCheckpointForRecovery(true);
 env.getCheckpointConfig().setTolerableCheckpointFailureNumber(0);
 
 // 3. 重启策略配置
 // 固定延迟重启
 env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000L));
 // 失败率重启
 env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.minutes(10), Time.minutes(1)));
 
 // socket文本流
 DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
 
 // 转换成SensorReading类型
 DataStream<SensorReading> dataStream = inputStream.map(line -> { 
 String[] fields = line.split(",");
 return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
 });
 
 dataStream.print();
 env.execute();
 }
 }

## Flink 的容错机制 ##

• 一致性检查点（checkpoint）  
• 从检查点恢复状态  
• Flink 检查点算法  
• 保存点（save points）

## 9.3 检查点（checkpoint） ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 45]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 46]  
**从检查点恢复状态**![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 47]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 48]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 49]  
![在这里插入图片描述][20210512174407308.png]

#### 检查点的实现算法 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 50]  
**Flink 检查点算法:**  
**核心思想和步骤：**  
**checkpoink线程发起checkpoint就是，往进入flink的数据流中插入一个特殊的数据barrier，source处理数据处理到checkpoink线程插入的barrier后就知道从barrier之前的数据做本次的状态的checkpoink保存, ，后面transition处理数据处理到checkpoink线程插入的barrier后就知  
道从barrier之前的数据做本次的计算状态的checkpoink保存, sink处理到checkpoink线程barrier也是从barrier之前的数据做本次的计算状态的checkpoink保存。**

**每次状态checkpoink都是保存到对应的barrier中。**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 51]

**source**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 52]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 53]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 54]  
**transition**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 55]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 56]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 57]  
**slink**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 58]

### 9.3.2 Flink+Kafka 如何实现端到端的 exactly-once 语义 ###

![在这里插入图片描述][20210512170925658.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 59]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 60]  
![在这里插入图片描述][20210512173844618.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 61]  
![在这里插入图片描述][20210512173919993.png]

## 9.4 保存点（Savepoints） ##

保存点和checkpoint很类似，如果没有开启checkpoint重启和升级时可以设置保存点，启动从保存点恢复之前计算的状态。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 62]

[20210512104315351.png]: /images/20221021/af206be6fb5e4aeba053c3d5f5032c8d.png
[20210512104514621.png]: /images/20221021/2e847c7dcd874cc5b28d9cadd4665012.png
[20210512104622629.png]: /images/20221021/6f49667a9c4e4b26a23971aa465020aa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70]: /images/20221021/7d7dfefdd03049bcb800e8fa9eaaac49.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 1]: /images/20221021/1639cbef559a45c4819d5f66054b721a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 2]: /images/20221021/038eaac7118b421aa4254c6b1329e03c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 3]: /images/20221021/c0688503bf184e7a94a999f121e7fb1f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 4]: /images/20221021/bc7b926240844903b7ccf86cd9398446.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 5]: /images/20221021/4c15f64af2394539b4adbaa9167419a0.png
[20210512104909438.png]: /images/20221021/bbf8294d3024493bb179f2bda187f2c2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 6]: /images/20221021/58cc42bcc08e4d34a0f83aea8f7bc917.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 7]: /images/20221021/5e3812ff755142929ea4bafe2bdabd75.png
[20210512105148796.png]: /images/20221021/2a66cf32215340f5b09e27839c0d54b4.png
[20210512105315706.png]: /images/20221021/668c86a102df4f57b6a30c52e4134a50.png
[20210512105452926.png]: /images/20221021/9064e8512da94f4abc02df3f41974bed.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 8]: /images/20221021/3cfcd8ada45a4a1aa248f4addd0cbfa4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 9]: /images/20221021/3a98c173e7df402a941820beb12966e4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 10]: /images/20221021/b2055f7187db43b09994cd43eeca82fa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 11]: /images/20221021/43742d4569694982b8608efd7f1e7ada.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 12]: /images/20221021/114def6ff66b4ea1961213406c7de3ef.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 13]: /images/20221021/20fb560cb4474689a654433b69f35d4d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 14]: /images/20221021/80b0c34e0df3406bbaa016f011b2ee8d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 15]: /images/20221021/7f8437d01b1544b48c3f08ea8de81923.png
[20210512143842849.png]: /images/20221021/88038d6f2d094a89adc7e7d871872ebf.png
[20210512143944573.png]: /images/20221021/aa696b1b489b47c8993a85cd09a6745c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 16]: /images/20221021/7befe9fd227645aa94c22e47bf273a23.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 17]: /images/20221021/2e5a987ee7b7442aaa979b5b9bc76489.png
[20210512144408989.png]: /images/20221021/926da0ac199347428112ab870bd8097a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 18]: /images/20221021/09bdc755c27844ddb16abc75ac657bed.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 19]: /images/20221021/6853440481a14bc3af04e06fbb1dee96.png
[2021051214584323.png]: /images/20221021/1f96eb6933a5431199023ab6b35d4c60.png
[20210512145801847.png]: /images/20221021/859b495548834d06899ee5810d0281d2.png
[20210512150500240.png]: /images/20221021/c38eac6ec5c14c40999aaaa37cdd3bc8.png
[20210512144720840.png]: /images/20221021/2e424c698c9b4981931c4caf81d15506.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 20]: /images/20221021/b75e93a781ff4deb8ae21c7e22a265c4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 21]: /images/20221021/d21c48c9047a4d40a499c4bddcbe6edc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 22]: /images/20221021/ad53bc8f0d9b4405b27d07c004700c8c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 23]: /images/20221021/16593d95f5f24ab394db680521e6351d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 24]: /images/20221021/a5ca4e7f895d445bb3b87b59efb05d4f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 25]: /images/20221021/b4d31981b49a4d6e9df9198b6fcf5b17.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 26]: /images/20221021/72a63857cc434f8bad7e8e808e4ae3df.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 27]: /images/20221021/1231b15051494da6a466a3fa798aac3b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 28]: /images/20221021/0981a8647755432cbdd3e899ee501b7f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 29]: /images/20221021/76d329ba6fd74f9b98cd5c20b083a73b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 30]: /images/20221021/b2131132477645bc9ece3db320cd63aa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 31]: /images/20221021/58937f0863f84b72a3d3bf4b4fa7360a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 32]: /images/20221021/9fac8ca4a5ee44a0a27a23708457895b.png
[20210512155103110.png]: /images/20221021/a833331ee3144ceab102d9ae3609db9f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 33]: /images/20221021/7746f9d6832241618626cdf1604c2177.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 34]: /images/20221021/2eb2adf085214e64a744c0b7209fd407.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 35]: /images/20221021/cd025b5618c54b55bc942666456a953a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 36]: /images/20221021/d082cc77e5e2410a9ec5bb45c324812f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 37]: /images/20221021/14c78d1ea7704aa2b834b52478126cd2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 38]: /images/20221021/7493b781803744f4849bf7ad7fd8c351.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 39]: /images/20221021/cdf92dc28f98423f91c9d6072a347d3a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 40]: /images/20221021/d9454630323849788e1625e2c072cf37.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 41]: /images/20221021/35c2f780ec5743ca83c1cc5d0891fdba.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 42]: /images/20221021/55bc72ae145f499693ab4c217d0ea5aa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 43]: /images/20221021/3f77850aa4ec411e8b04dedf0426bb97.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 44]: /images/20221021/5d953a80a7994286a34c9bc84eeff37b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 45]: /images/20221021/9a6149f159d445c5a0c029435f4f742e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 46]: /images/20221021/10b69c5c1bf24591a13b0e0031a4a85d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 47]: /images/20221021/2fafa10e096c47859894d1da3849ba6b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 48]: /images/20221021/1996e8474b38430bbb4ec00985dbed01.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 49]: /images/20221021/14a1c18195844397a1884d455fdec58d.png
[20210512174407308.png]: /images/20221021/039f86f3a6fe445faa29db5f6675b2e9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 50]: /images/20221021/e27fcde9806743b49d261bab5f6370e9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 51]: /images/20221021/2121dcfaf4e340f4a1ea41f7e0909085.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 52]: /images/20221021/ac91d4d204d94696a4e4f755d777f44b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 53]: /images/20221021/6e86ead058024166bc19fe31971f8206.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 54]: /images/20221021/04d138f67ff94bef814478cda67b1f9e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 55]: /images/20221021/44530ed70d2b4514b36df0c59cf47fce.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 56]: /images/20221021/93a270083aa14b26bac45ef49ebf31f9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 57]: /images/20221021/e79a054a8b5c4036a8f34777aae84263.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 58]: /images/20221021/c1f1c5d1a1cd4d44a783ebea575be2a1.png
[20210512170925658.png]: /images/20221021/57ae5a4864fe4384b8db8388916142a9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 59]: /images/20221021/e6e92d5d63dd4d77b52e7151ec6e01ea.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 60]: /images/20221021/68442b02e5c4423f8271ced3b0e0ea51.png
[20210512173844618.png]: /images/20221021/719b55018a974523a367f08477c893b4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 61]: /images/20221021/8cb9488255944ffa860118b9548e9ed4.png
[20210512173919993.png]: /images/20221021/6f4ac18ab6a14bbf81f1e0c0102f96cd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbmdzaGVuZ3dlaTIzMDYxMg_size_16_color_FFFFFF_t_70 62]: /images/20221021/8f063aac167e45b0be0d1dbb76b661de.png