Flink状态管理及状态后端配置

冷不防 2022-11-06 00:59 519阅读 0赞

# Flink状态管理及状态后端配置 #

### 目录 ###

*  Flink状态管理及状态后端配置
 *   *  1. 什么是状态
     *  2. 为什么需要管理状态
     *   *  3. Flink中的状态分类
     *  4. Managed State的分类
     *  5. 算子状态的使用
     *   *  案例1: 列表状态
         *  案例2: 广播状态
         *  6. 键控状态的使用
         *  案例1: ValueState
         *  案例2: ListState
         *  案例3: ReducingState
         *  案例4: AggregatingState
         *  案例5:MapState
     *  7. 状态后端
     *   *  状态后端的分类
         *  配置状态后端

`有状态的计算`是`流处理框架`要实现的重要功能，因为稍复杂的流处理场景都需要记录状态，然后在新流入数据的基础上不断更新状态。  
SparkStreaming在状态管理这块做的不好, 很多时候需要借助于外部存储(例如Redis)来手动管理状态, 增加了编程的难度.  
`Flink的状态管理是它的优势之一.`

## 1. 什么是状态 ##

在流式计算中有些操作一次处理一个独立的事件(比如解析一个事件), 有些操作却需要记住多个事件的信息(比如窗口操作).  
那些需要记住多个事件信息的操作就是`有状态`的.  
流式计算分为`无状态计算和有状态计算`两种情况。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]

无状态的计算`观察每个独立事件`，并根据最后一个事件输出结果。例如，流处理应用程序从传感器接收水位数据，并在水位超过指定高度时发出警告。

有状态的计算则会`基于多个事件输出结果`。以下是一些例子。例如，计算过去一小时的平均水位，就是有状态的计算。所有用于复杂事件处理的状态机。例如，若在一分钟内收到两个相差20cm以上的水位差读数，则发出警告，这是有状态的计算。流与流之间的所有关联操作，以及流与静态表或动态表之间的关联操作，都是有状态的计算。

## 2. 为什么需要管理状态 ##

下面的几个场景都需要使用流处理的状态功能:

*  **去重**  
    数据流中的数据有重复，我们想对重复数据去重，需要记录哪些数据已经流入过应用，当新数据流入时，根据已流入过的数据来判断去重。
 *  **检测**  
    检查输入流是否符合某个特定的模式，需要将之前流入的元素以状态的形式缓存下来。比如，判断一个温度传感器数据流中的温度是否在持续上升。
 *  **聚合**  
    对一个时间窗口内的数据进行聚合分析，分析一个小时内水位的情况
 *  **更新机器学习模型**  
    在线机器学习场景下，需要根据新流入数据不断更新机器学习的模型参数。

### 3. Flink中的状态分类 ###

Flink包括两种基本类型的状态`Managed State`和`Raw State`

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th>Managed State</th> 
   <th>Raw State</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>状态管理方式</td> 
   <td>Flink Runtime托管, 自动存储, 自动恢复, 自动伸缩</td> 
   <td>用户自己管理</td> 
  </tr> 
  <tr> 
   <td>状态数据结构</td> 
   <td>Flink提供多种常用数据结构, 例如:ListState, MapState等</td> 
   <td>字节数组: byte[]</td> 
  </tr> 
  <tr> 
   <td>使用场景</td> 
   <td>绝大数Flink算子</td> 
   <td>所有算子</td> 
  </tr> 
 </tbody> 
</table>

**注意**:  
从具体使用场景来说，绝大多数的算子都可以通过继承Rich函数类或其他提供好的接口类，在里面使用Managed State。Raw State一般是在已有算子和Managed State不够用时，用户自定义算子时使用。  
在我们平时的使用中Managed State已经足够我们使用.

## 4. Managed State的分类 ##

对Managed State继续细分，它又有两种类型

*  Operator State(算子状态)
 *  Keyed State(键控状态)

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th>Operator State</th> 
   <th>Keyed State</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>适用用算子类型</td> 
   <td>可用于所有算子: <code>常用于source</code>, 例如 FlinkKafkaConsumer</td> 
   <td>只适用于KeyedStream上的算子</td> 
  </tr> 
  <tr> 
   <td>状态分配</td> 
   <td>一个算子的<code>子任务</code>对应<code>一个状态</code></td> 
   <td>一个Key对应一个State: 一个算子会处理多个Key, 则访问相应的多个State</td> 
  </tr> 
  <tr> 
   <td>创建和访问方式</td> 
   <td>实现CheckpointedFunction或ListCheckpointed(<code>已经过时</code>)接口</td> 
   <td>重写RichFunction, 通过里面的RuntimeContext访问</td> 
  </tr> 
  <tr> 
   <td>横向扩展</td> 
   <td>并发改变时有多重重写分配方式可选: 均匀分配和合并后每个得到全量</td> 
   <td>发改变, State随着Key在实例间迁移</td> 
  </tr> 
  <tr> 
   <td>支持的数据结构</td> 
   <td>ListState和BroadCastState</td> 
   <td>ValueState, ListState,MapState ReduceState, AggregatingState</td> 
  </tr> 
 </tbody> 
</table>

## 5. 算子状态的使用 ##

Operator State可以用在`所有算子`上，每个算子子任务或者说每个算子实例共享一个状态，流入这个算子子任务的数据可以访问和更新这个状态。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]

注意: 算子`子任务(并行度)`之间的状态不能互相访问

Operator State的实际应用场景不如Keyed State多，它经常`被用在Source或Sink等算子上`，用来保存流入数据的偏移量或对输出数据做缓存，以保证Flink应用的`Exactly-Once`语义。

**Flink为算子状态提供三种基本数据结构**：

*  列表状态（List state）  
    将状态表示为一组数据的列表
 *  联合列表状态（Union list state）  
    也是将状态表示为数据的列表。它与常规列表状态的区别在于，在发生故障时，或者从保存点（savepoint）启动应用程序时如何恢复。  
    一种是均匀分配(List state)，另外一种是将所有 State 合并为全量 State 再分发给每个实例(Union list state)。
 *  广播状态（Broadcast state）  
    是一种特殊的算子状态. 如果一个算子有多项任务，而它的每项任务状态又都相同，那么这种特殊情况最适合应用广播状态。

### 案例1: 列表状态 ###

在map算子中计算数据的个数

import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.common.state.ListState;
    import org.apache.flink.api.common.state.ListStateDescriptor;
    import org.apache.flink.runtime.state.FunctionInitializationContext;
    import org.apache.flink.runtime.state.FunctionSnapshotContext;
    import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    public class S01_Operator_List { 
        public static void main(String[] args) { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(3);
    
            env.socketTextStream("hadoop162",9999)
                    .map(new MyMapFunction())
                    .print();
    
            try { 
                env.execute();
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    
        public static class MyMapFunction implements MapFunction<String,String>, CheckpointedFunction{ 
    
            long count =0;
            private ListState<Long> state;
    
            @Override
            public String map(String value) throws Exception { 
                count++;
                state.clear();
                state.add(count);
                return count+"";
            }
    
            //初始化算子状态
            @Override
            public void initializeState(FunctionInitializationContext context) throws Exception { 
                //System.out.println("initializeState ...... ");
                //state = context.getOperatorStateStore().getListState(new ListStateDescriptor<Long>("state",Long.class));
                state = context.getOperatorStateStore()
                        .getUnionListState(new ListStateDescriptor<Long>("state",Long.class));
    
                Iterator<Long> it = state.get().iterator();
                if (it.hasNext()){ 
                    count = it.next();
                }
            }
    
            //checkpoint时调用这个方法，在这里可以写持久化。
            @Override
            public void snapshotState(FunctionSnapshotContext context) throws Exception { 
                //System.out.println("snapshotState ...... ");
                state.clear();
                state.add(count);
            }
        }
    }

### 案例2: 广播状态 ###

从版本1.5.0开始，Apache Flink具有一种新的状态，称为`广播状态`。  
广播状态被引入以支持这样的用例:来自一个流的一些数据需要广播到所有下游任务，在那里它被本地存储，并用于处理另一个流上的所有传入元素。作为广播状态自然适合出现的一个例子，我们可以想象一个低吞吐量流，其中包含一组规则，我们希望根据来自另一个流的所有元素对这些规则进行评估。考虑到上述类型的用例，广播状态与其他操作符状态的区别在于:

1.  它是一个map格式
2.  它只对输入有广播流和无广播流的特定操作符可用
3.  这样的操作符可以具有不同名称的多个广播状态。

import org.apache.flink.api.common.state.BroadcastState;
    import org.apache.flink.api.common.state.MapStateDescriptor;
    import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
    import org.apache.flink.streaming.api.datastream.BroadcastStream;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
    import org.apache.flink.util.Collector;
    
    ublic class S02_Operator_Broadcast { 
        public static void main(String[] args) { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(3);
            env.enableCheckpointing(1000) ;
    
            //数据流
            DataStreamSource<String> dataStream = env.socketTextStream("hadoop162", 9999);
            //控制流
            BroadcastStream<String> controlStream = env.socketTextStream("hadoop162", 8888)
                    .broadcast(new MapStateDescriptor<String, String>("bd_state", String.class, String.class));
    
            dataStream.connect(controlStream)
                    .process(new BroadcastProcessFunction<String, String, String>() { 
                        @Override
                        public void processElement(String value,ReadOnlyContext ctx,Collector<String> out) throws Exception { 
                            // 从广播状态中取值, 不同的值做不同的处理
                            ReadOnlyBroadcastState<String, String> state =
                                    ctx.getBroadcastState(new MapStateDescriptor<>("bd_state", String.class, String.class));
    
                            String seitch = state.get("switch");
                            seitch = seitch==null?"0" : seitch;
                            switch (seitch){ 
                                case "1":
                                    out.collect("进入一级战备状态");
                                     break;
                                case "2":
                                    out.collect("进入二级战备状态");
                                    break;
                                default:
                                    out.collect("平常式三级戒备");
                                    break;
                            }
                        }
    
                        @Override
                        public void processBroadcastElement(String value,Context ctx,Collector<String> out) throws Exception { 
                            //把值放入广播状态
                            ctx.getBroadcastState(new MapStateDescriptor<>("bd_state",String.class,String.class))
                                    .put("switch",value);
                        }
                    })
                    .print();
    
            try { 
                env.execute();
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }

### 6. 键控状态的使用 ###

键控状态是根据输入数据流中定义的键（key）来维护和访问的。  
Flink为每个键值维护一个状态实例，并将具有相同键的所有数据，都分区到同一个算子任务中，这个任务会维护和处理这个key对应的状态。当任务处理一条数据时，它会自动将状态的访问范围限定为当前数据的key。因此，`具有相同key的所有数据都会访问相同的状态`。  
Keyed State很类似于一个分布式的key-value map数据结构，`只能用于KeyedStream（keyBy算子处理之后）`。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 2]

**键控状态支持的数据类型**

*  ValueState  
    保存单个值. 每个有key有一个状态值. 设置使用 update(T), 获取使用 T value()
 *  ListState:  
    保存元素列表.  
    添加元素: add(T) addAll(List)  
    获取元素: Iterable get()  
    **覆盖**所有元素: update(List)
 *  ReducingState:  
    存储单个值, 表示把所有元素的聚合结果添加到状态中. 与ListState类似, 但是当使用add(T)的时候ReducingState会使用指定的**ReduceFunction**进行聚合.
 *  AggregatingState<IN, OUT>:  
    存储单个值. 与ReducingState类似, 都是进行聚合. 不同的是, AggregatingState的聚合的结果和元素类型可以不一样.
 *  MapState<UK, UV>:  
    存储键值对列表.  
    添加键值对: put(UK, UV) or putAll(Map<UK, UV>)  
    根据key获取值: get(UK)  
    获取所有: entries(), keys() and values()  
    检测是否为空: isEmpty()

**注意**:

1.  所有的类型都有clear(), 清空当前key的状态
2.  这些状态对象仅用于用户与状态进行交互.
3.  状态不是必须存储到内存, 也可以存储在磁盘或者任意其他地方
4.  从状态获取的值与输入元素的key相关

### 案例1: ValueState ###

检测传感器的水位值，如果连续的两个水位值超过10，就输出报警。

import com.atguigu.flink.java.chapter_5.WaterSensor;
    import org.apache.flink.api.common.state.ValueState;
    import org.apache.flink.api.common.state.ValueStateDescriptor;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
    import org.apache.flink.util.Collector;
    
    public class S03_Keyed_Value { 
        public static void main(String[] args) throws IOException { 
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(3);
    
            env.socketTextStream("hadoop162",9999)
                    .map(value->{ 
                        String[] split = value.split(",");
                        return new WaterSensor(split[0],Long.valueOf(split[1]),Integer.valueOf(split[2]));
                    })
                    .keyBy(WaterSensor::getId)
                    .process(new KeyedProcessFunction<String, WaterSensor, String>() { 
                        private ValueState<Integer> state;
                        
                        @Override
                        public void open(Configuration parameters) throws Exception { 
                            //初始化状态
                            state = getRuntimeContext()
                                    .getState(new ValueStateDescriptor<Integer>("state", Integer.class));
                        }
    
                        //连续两次超警戒水位
                        @Override
                        public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception { 
                            Integer lastVc = state.value();
                            if (lastVc != null){ 
                                if (value.getVc()>10 && lastVc >10){ 
                                    out.collect(value.getId() + "连续两次超警戒水位");
                                }
                            }
                            state.update(value.getVc());
                        }
                    })
                    .print();
    
            try { 
                env.execute();
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }

### 案例2: ListState ###

针对每个传感器输出最高的3个水位值

import com.atguigu.flink.java.chapter_5.WaterSensor;
    import org.apache.flink.api.common.state.ListState;
    import org.apache.flink.api.common.state.ListStateDescriptor;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
    import org.apache.flink.util.Collector;
    import java.util.ArrayList;
    import java.util.List;
    
    public class S03_Keyed_List { 
        public static void main(String[] args) throws IOException { 
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(3);
    
            env.socketTextStream("hadoop162",9999)
                    .map(value->{ 
                        String[] split = value.split(",");
                        return new WaterSensor(split[0],Long.valueOf(split[1]),Integer.valueOf(split[2]));
                    })
                    .keyBy(WaterSensor::getId)
                    .process(new KeyedProcessFunction<String, WaterSensor, String>() { 
                        private ListState<Integer> state;
    
                        @Override
                        public void open(Configuration parameters) throws Exception { 
                            //初始化状态
                            state = getRuntimeContext().getListState(new ListStateDescriptor<Integer>("state", Integer.class));
                        }
    
                        //排序取前三大的水位
                        @Override
                        public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception { 
                            state.add(value.getVc()); //个数小于等于4个，利用减一的方式取出排序后数组的前三名
    
                            ArrayList<Integer> integers = new ArrayList<>();
                            for (Integer integer : state.get()) { 
                                integers.add(integer);
                            }
                            //排序
                            integers.sort((o1, o2) -> o2 - o1);
                            //取前三
                            if (integers.size() > 3){ 
                                integers.remove(3);
                            }
                            state.update(integers);
    
                            //out.collect(value.getId() + ":" + state.get());
                            out.collect(ctx.getCurrentKey() +": "+integers);
                        }
                    })
                    .print();
    
            try { 
                env.execute();
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }

### 案例3: ReducingState ###

计算每个传感器的水位和

.process(new KeyedProcessFunction<String, WaterSensor, String>() {
                private ReducingState<Integer> state;
    
                @Override
                public void open(Configuration parameters) throws Exception {
                    //初始化状态
                    state = getRuntimeContext()
                            .getReducingState(new ReducingStateDescriptor<Integer>("state",
                                            new ReduceFunction<Integer>() {
                                                @Override
                                                public Integer reduce(Integer value1, Integer value2) throws Exception {
                                                    return value1 + value2;
                                                }
                                            },
                                            Integer.class));
                }
                
                @Override
                public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                    state.add(value.getVc());
                    out.collect(value.getId() + ":" + state.get());
                }
            })
            .print();

### 案例4: AggregatingState ###

计算每个传感器的平均水位

.process(new KeyedProcessFunction<String, WaterSensor, Double>() { 
    
        private AggregatingState<Integer, Double> avgState;
    
        @Override
        public void open(Configuration parameters) throws Exception { 
            AggregatingStateDescriptor<Integer, Tuple2<Integer, Integer>, Double> aggregatingStateDescriptor = new AggregatingStateDescriptor<>("avgState", new AggregateFunction<Integer, Tuple2<Integer, Integer>, Double>() { 
                @Override
                public Tuple2<Integer, Integer> createAccumulator() { 
                    return Tuple2.of(0, 0);
                }
    
                @Override
                public Tuple2<Integer, Integer> add(Integer value, Tuple2<Integer, Integer> accumulator) { 
                    return Tuple2.of(accumulator.f0 + value, accumulator.f1 + 1);
                }
    
                @Override
                public Double getResult(Tuple2<Integer, Integer> accumulator) { 
                    return accumulator.f0 * 1D / accumulator.f1;
                }
    
                @Override
                public Tuple2<Integer, Integer> merge(Tuple2<Integer, Integer> a, Tuple2<Integer, Integer> b) { 
                    return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);
                }
            }, Types.TUPLE(Types.INT, Types.INT));
            avgState = getRuntimeContext().getAggregatingState(aggregatingStateDescriptor);
        }
    
        @Override
        public void processElement(WaterSensor value, Context ctx, Collector<Double> out) throws Exception { 
            avgState.add(value.getVc());
            out.collect(avgState.get());
        }
    })

### 案例5:MapState ###

去重: 去掉重复的水位值. 思路: 把水位值作为MapState的key来实现去重, value随意

env.socketTextStream("hadoop162",9999)
                    .map(value->{ 
                        String[] split = value.split(",");
                        return new WaterSensor(split[0],Long.valueOf(split[1]),Integer.valueOf(split[2]));
                    })
                    .keyBy(WaterSensor::getId)
                    .process(new KeyedProcessFunction<String, WaterSensor, String>() {        
                        private MapState<Integer, Object> state;
                        
                        @Override
                        public void open(Configuration parameters) throws Exception { 
                            //初始化状态
                            state = getRuntimeContext().getMapState(new MapStateDescriptor<Integer, Object>("state", Integer.class, Object.class));
                        }
    
                        @Override
                        public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception { 
                            state.put(value.getVc(), new Object());
                            out.collect(value.getId() + ":" + state.keys());
                        }
                    })
                    .print();

## 7. 状态后端 ##

每传入一条数据，有状态的算子任务都会`读取和更新状态`。由于有效的状态访问对于处理数据的低延迟至关重要，因此每个并行任务(子任务)都会在本地维护其状态，以确保快速的状态访问。  
状态的存储、访问以及维护，由一个`可插入`的组件决定，这个组件就叫做`状态后端`（state backend）  
`状态后端主要负责两件事`：

*  本地(taskmanager)的状态管理
 *  将检查点（checkpoint）状态写入远程存储

### 状态后端的分类 ###

状态后端作为一个可插入的组件, 没有固定的配置, 我们可以根据需要选择一个合适的状态后端.  
Flink提供了3种状态后端:

*  MemoryStateBackend  
    内存级别的状态后端,  
    **存储方式**:本地状态存储在TaskManager的内存中, checkpoint 存储在JobManager的内存中.  
    **特点**:快速, 低延迟, 但不稳定  
    **使用场景**: 1. 本地测试 2. 几乎无状态的作业(ETL) 3. JobManager不容易挂, 或者挂了影响不大. 4. 不推荐在生产环境下使用
 *  FsStateBackend  
    **存储方式**: 本地状态在TaskManager内存, `Checkpoint时, 先把状态发送到JobManager的内存, 然后再存储在文件系统中`  
    **特点**: 拥有内存级别的本地访问速度, 和更好的容错保证  
    **使用场景**: 1. 常规使用状态的作业. 例如分钟级别窗口聚合, join等 2. 需要开启HA的作业 3. 可以应用在生产环境中
 *  RocksDBStateBackend  
    将所有的状态序列化之后, 存入本地的`RocksDB`数据库中.(一种NoSql数据库, KV形式存储)  
    **存储方式**: 1. 本地状态存储在TaskManager的RocksDB数据库中(实际是内存+磁盘) 2. Checkpoint在外部文件系统中.  
    **使用场景**: 1. 超大状态的作业, 例如天级的窗口聚合 2. 需要开启HA的作业 3. 对读写状态性能要求不高的作业 4. 可以使用在生产环境

### 配置状态后端 ###

**全局配置状态后端**  
在flink-conf.yaml文件中设置默认的全局后端  
![在这里插入图片描述][20210307131626578.png]

**在代码中配置状态后端**  
可以在代码中单独为这个Job设置状态后端.

env.setStateBackend(new MemoryStateBackend());
    env.setStateBackend(new FsStateBackend("hdfs://hadoop162:8020/flink/checkpoints/fs"));

如果要使用RocksDBBackend, 需要先引入依赖:

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

env.setStateBackend(new RocksDBStateBackend("hdfs://hadoop162:8020/flink/checkpoints/rocksdb"));

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221023/1337393f168e4effab9fdc69156fd499.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]: /images/20221023/c93cccdd104c43518d8efad65f67ce4e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 2]: /images/20221023/5ddd17d88b7d4dbe906a22d91cadb226.png
[20210307131626578.png]: /images/20221023/3c230603bfa3449dac1bafeecbfdb6ea.png