Flink电商指标分析项目(2)_Flink整合kafka开发

电玩女神 2022-11-06 00:59 265阅读 0赞

### 目录 ###

*   *  Flink实时数据分析系统开发
     *  搭建【Flink实时数据分析系统】项目环境
     *   *  导入Maven项目依赖
         *  创建项目包结构
         *  导入实时系统Kafka/Hbase配置
         *  获取配置文件API介绍
     *  初始化Flink流式计算环境
     *  Flink添加checkpoint容错支持
     *  Flink整合Kafka
     *   *  Flink读取Kafa数据
         *  Kafka消息解析为元组
         *  Flink封装点击流消息为样例类
         *  封装Kafka消息为Message样例类
     *  Flink添加水印支持

## Flink实时数据分析系统开发 ##

前边我们已经开发完毕了`上报服务系统`, 我们可以通过上报服务系统把电商页面中的点击流数据发送到Kafka中, 那么接下来我们就来开发`Flink实时分析系统`, 通过流的方式读取Kafka中的消息, 进而分析数据。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]  
**业务**

*  实时分析频道热点
 *  实时分析频道PV/UV
 *  实时分析频道新鲜度
 *  实时分析频道地域分布
 *  实时分析运营商平台
 *  实时分析浏览器类型  
    **技术**
 *  Flink实时处理算子
 *  使用`CheckPoint`和`水印`解决Flink生产上遇到的问题（网络延迟、丢数据）
 *  Flink整合Kafka
 *  Flink整合HBase

## 搭建【Flink实时数据分析系统】项目环境 ##

### 导入Maven项目依赖 ###

1.  将pom.xml文件中的依赖导入到`real-process`项目的pom.xml

<properties>
            <scala.version>2.11</scala.version>
            <flink.version>1.6.1</flink.version>
            <hadoop.version>2.7.5</hadoop.version>
            <hbase.version>2.0.0</hbase.version>
        </properties>
    
    
        <dependencies>
            
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka_${ scala.version}</artifactId>
                <version>0.10.1.0</version>
            </dependency>
    
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-connector-kafka-0.10_${ scala.version}</artifactId>
                <version>${ flink.version}</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table_${ scala.version}</artifactId>
                <version>${ flink.version}</version>
            </dependency>
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-scala_${ scala.version}</artifactId>
                <version>${ flink.version}</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-scala_${ scala.version}</artifactId>
                <version>${ flink.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java_${ scala.version}</artifactId>
                <version>${ flink.version}</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-hbase_${ scala.version}</artifactId>
                <version>${ flink.version}</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.hbase</groupId>
                <artifactId>hbase-client</artifactId>
                <version>${ hbase.version}</version>
            </dependency>
    
    
            
    
    
            
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-common</artifactId>
                <version>${ hadoop.version}</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-hdfs</artifactId>
                <version>${ hadoop.version}</version>
                
                <exclusions>
                    <exclusion>
                        <groupId>xml-apis</groupId>
                        <artifactId>xml-apis</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
    
            <dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-client</artifactId>
                <version>${ hadoop.version}</version>
                
                <exclusions>
                    <exclusion>
                        <groupId>com.google.protobuf</groupId>
                        <artifactId>protobuf-java</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
    
    
            
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.44</version>
            </dependency>
        </dependencies>
    
    
        <build>
            <sourceDirectory>src/main/scala</sourceDirectory>
            <testSourceDirectory>src/test/scala</testSourceDirectory>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-shade-plugin</artifactId>
                    <version>3.0.0</version>
                    <executions>
                        <execution>
                            <phase>package</phase>
                            <goals>
                                <goal>shade</goal>
                            </goals>
                            <configuration>
                                <artifactSet>
                                    <excludes>
                                        <exclude>com.google.code.findbugs:jsr305</exclude>
                                        <exclude>org.slf4j:*</exclude>
                                        <exclude>log4j:*</exclude>
                                    </excludes>
                                </artifactSet>
                                <filters>
                                    <filter>
                                        
                                        <artifact>*:*</artifact>
                                        <excludes>
                                            <exclude>META-INF/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> </excludes> </filter> </filters> <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> <mainClass>com.itheima.realprocess.App</mainClass> </transformer> </transformers> </configuration> </execution> </executions> </plugin> </plugins> </build>

1.  `real-process`模块添加scala支持
2.  main和test创建`scala`文件夹，并标记为源代码和测试代码目录

### 创建项目包结构 ###

<table> 
 <thead> 
  <tr> 
   <th>包名</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>com.itheima.realprocess.util</td> 
   <td>存放工具类</td> 
  </tr> 
  <tr> 
   <td>com.itheima.realprocess.bean</td> 
   <td>存放实体类</td> 
  </tr> 
  <tr> 
   <td>com.itheima.realprocess.task</td> 
   <td>存放具体的分析任务<br>每一个业务都是一个任务，对应的分析处理都写在这里</td> 
  </tr> 
 </tbody> 
</table>

### 导入实时系统Kafka/Hbase配置 ###

1.  application.conf`导入到`resources\`目录

#
    #kafka的配置
    #
    # Kafka集群地址
    bootstrap.servers="hadoop102:9092,hadoop103:9092,hadoop104:9092"
    # ZooKeeper集群地址
    zookeeper.connect="hadoop102:2181,hadoop103:2181,hadoop104:2181"
    # Kafka Topic名称
    input.topic="pyg"
    # 消费组ID
    group.id="pyg"
    # 自动提交拉取到消费端的消息offset到kafka
    enable.auto.commit="true"
    # 自动提交offset到zookeeper的时间间隔单位（毫秒）
    auto.commit.interval.ms="5000"
    # 每次消费最新的数据
    auto.offset.reset="latest"

1.  `log4j.properties`到`resources`目录下

log4j.rootLogger=warn,stdout
    log4j.appender.stdout=org.apache.log4j.ConsoleAppender 
    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout 
    log4j.appender.stdout.layout.ConversionPattern=%5p - %m%n

注意修改`kafka服务器`和`hbase服务器`的机器名称

### 获取配置文件API介绍 ###

**ConfigFactory.load()介绍**

*  使用`ConfigFactory.load()`可以自动加载配置文件中的`application.conf`文件（`注意：名字一定不要写错，否则无法加载`），并返回一个Config对象
 *  使用Config对象可以获取到配置文件中的配置项
 *  `application.conf`文件是一个properties文件，存放key-value键值对的数据。  
    **常用API**

<table> 
 <thead> 
  <tr> 
   <th>方法名</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>getString(“key”)</td> 
   <td>获取配置文件中指定key的值对应的字符串</td> 
  </tr> 
  <tr> 
   <td>getInt(“key”)</td> 
   <td>获取配置文件中指定key的值对应的整型数字</td> 
  </tr> 
  <tr> 
   <td>getLong(“key”)</td> 
   <td>同上</td> 
  </tr> 
  <tr> 
   <td>getBoolean(“key”)</td> 
   <td>同上</td> 
  </tr> 
 </tbody> 
</table>

\#\#3 编写scala代码读取配置工具类  
在`com.itheima.realprocess.util`包下创建`GlobalConfigUtil`单例对象（object）  
**步骤**

1.  使用`ConfigFactory.load`获取配置对象
2.  编写方法加载`application.conf`配置
3.  添加一个`main`方法测试，工具类是否能够正确读取出配置项。  
    **示例代码：**  
    **GlobalConfigUtil.scala**

/** * 加载配置文件工具类 */
    object GlobalConfigUtil { 
    
      private val config: Config = ConfigFactory.load()
    
      /*------------------ * Kafka配置 *------------------*/
    
      val bootstrapServers = config.getString("bootstrap.servers")
      val zookeeperConnect = config.getString("zookeeper.connect")
      val inputTopic = config.getString("input.topic")
      val groupId = config.getString("group.id")
      val enableAutoCommit = config.getString("enable.auto.commit")
      val autoCommitIntervalMs = config.getString("auto.commit.interval.ms")
      val autoOffsetReset = config.getString("auto.offset.reset")
        
      
      // 测试配置文件读取类
      def main(args: Array[String]): Unit = { 
        println(bootstrapServers)
        println(zookeeperConnect)
        println(inputTopic)
        println(groupId)
        println(enableAutoCommit)
        println(autoCommitIntervalMs)
        println(autoOffsetReset)
      }
    }

## 初始化Flink流式计算环境 ##

**步骤**

1.  创建`App`单例对象
2.  创建main方法，获取`StreamExecutionEnvironment`运行环境
3.  设置流处理的时间为`EventTime`，使用数据发生的时间来进行数据处理
4.  设置Flink的并行度
5.  编写测试代码，测试Flink程序是否能够正确执行  
    **参考代码**

import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.api.scala._
    
    object App { 
      def main(args: Array[String]): Unit = { 
    
        //初始化Flink的流式环境
        val env = StreamExecutionEnvironment.getExecutionEnvironment
    
        //设置处理的时间为EventTime
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
        //设置并行度
        env.setParallelism(1)
    
        //本地测试 加载本地集合 成为一个DataStream 打印输出
        val localDataStream = env.fromCollection(
          List("hadoop", "hive", "hbase", "flink")
        )
    
        localDataStream.print()
    
        //执行任务
        env.execute("real-process")
      }
    
    }

注意：
    1. 一定要导入`import org.apache.flink.api.scala._`隐式转换，否则Flink程序无法执行
    2. 到导入`org.apache.flink.streaming.api`下的`TimeCharacteristic`，否则`没有EventTime`

## Flink添加checkpoint容错支持 ##

`Checkpoint`是Flink实现容错机制最核心的功能，它能够根据配置周期性地基于Stream中各个Operator的状态来生成`Snapshot`，从而将这些状态数据定期持久化存储下来，当Flink程序一旦意外崩溃时，重新运行程序时可以有选择地从这些Snapshot进行恢复，从而修正因为故障带来的程序数据状态中断。  
**步骤**

1.  Flink envrionment中添加`checkpoint`支持
2.  运行Flink程序测试checkpoint是否配置成功（检查HDFS中是否已经保存snapshot数据）  
    **实现**  
    在Flink流式处理环境中，添加以下checkpoint的支持，确保Flink的高容错性，数据不丢失。

//
    // 保证程序长时间运行的安全性进行checkpoint操作
    //
    // 5秒启动一次checkpoint
    env.enableCheckpointing(5000)
    // 设置checkpoint只checkpoint一次
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    // 设置两次checkpoint的最小时间间隔
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(1000)
    // checkpoint超时的时长
    env.getCheckpointConfig.setCheckpointTimeout(60000)
    // 允许的最大checkpoint并行度
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
    // 当程序关闭的时，触发额外的checkpoint
    env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    // 设置checkpoint的地址
    env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flink-checkpoint/"))

1.  启动`HDFS`
2.  启动`Flink程序`测试
3.  如果测试成功，在HDFS中应该生成如下几个目录

## Flink整合Kafka ##

### Flink读取Kafa数据 ###

**步骤**

1.  配置Kafka连接属性
2.  使用`FlinkKafkaConsumer010`整合Kafka
3.  添加一个source到当前Flink环境
4.  启动`zookeeper`
5.  启动`kafka`
6.  运行Flink程序测试是否能够从Kafka中消费到数据  
    **实现**

1.配置Kafka连接属性

//
    // 整合Kafka
    //
    val properties = new Properties()
    
    // # Kafka集群地址
    properties.setProperty("bootstrap.servers",GlobalConfigUtil.bootstrapServers)
    // # ZooKeeper集群地址
    properties.setProperty("zookeeper.connect",GlobalConfigUtil.zookeeperConnect)
    // # Kafka Topic名称
    properties.setProperty("input.topic",GlobalConfigUtil.inputTopic)
    // # 消费组ID
    properties.setProperty("group.id",GlobalConfigUtil.groupId)
    // # 自动提交拉取到消费端的消息offset到kafka
    properties.setProperty("enable.auto.commit",GlobalConfigUtil.enableAutoCommit)
    // # 自动提交offset到zookeeper的时间间隔单位（毫秒）
    properties.setProperty("auto.commit.interval.ms",GlobalConfigUtil.autoCommitIntervalMs)
    // # 每次消费最新的数据
    properties.setProperty("auto.offset.reset",GlobalConfigUtil.autoOffsetReset)
    
    val consumer = new FlinkKafkaConsumer010[String](
        GlobalConfigUtil.inputTopic,
        new SimpleStringSchema(),
        properties
    )

2.添加一个source到当前Flink环境

val kafkaDataStream: DataStream[String] = env.addSource(consumer)

3.打印DataStream中的数据

kafkaDataStream.print()

1.  启动`zookeeper`
2.  启动`kafka`
3.  运行Flink程序
4.  运行上报服务系统
5.  启动消息生成器, 测试是否能够从Kafka中消费到数据  
    **如果Flink从Kafka消费成功会打印以下数据，就证明我们的代码是正确的。**

{ 
          "count": 1,
          "message": "{\"browserType\":\"谷歌浏览器\",\"categoryID\":6,\"channelID\":4,\"city\":\"America\",\"country\":\"china\",\"entryTime\":1544601660000,\"leaveTime\":1544634060000,\"network\":\"联通\",\"produceID\":4,\"province\":\"china\",\"source\":\"百度跳转\",\"userID\":13}",
          "timeStamp": 1553188417573
      }

### Kafka消息解析为元组 ###

**步骤**

*  使用map算子，遍历kafka中消费到的数据
 *  使用FastJSON转换为JSON对象
 *  将JSON的数据解析成一个元组
 *  打印map映射后的元组数据
 *  测试是否能够正确解析  
    **代码**

1.  使用map算子，将kafka中消费到的数据，使用FastJSON准换为JSON对象
2.  将JSON的数据解析成一个元组

object App { 
      def main(args: Array[String]): Unit = { 
        ...
          
        //JSON->元组
        val tupleDataStream = kafkaDataStream.map { 
          msgJSON => { 
            val jsonObject = JSON.parseObject(msgJSON)
    
            val message = jsonObject.getString("message")
            val count = jsonObject.getLong("count")
            val timeStamp = jsonObject.getLong("timeStamp")
    
            (message, count, timeStamp)
          }
        }
        tupleDataStream.print()
        
        //执行任务
        env.execute("real-process")
      }
    
    }

1.  打印经过map映射后的元组数据，测试是否能够正确解析  
    原来的格式  
    ![在这里插入图片描述][20210307150410957.png]  
    解析后的格式  
    ![在这里插入图片描述][20210307150520284.png]

### Flink封装点击流消息为样例类 ###

**步骤**

1.  创建一个`ClickLog`样例类中来封装消息
2.  使用map算子将数据封装到`ClickLog`样例类  
    **代码**
3.  在bean包中，创建`ClickLog`样例类，添加以下字段

*  频道ID(channelID)
 *  产品类别ID(categoryID)
 *  产品ID(produceID)
 *  国家(country)
 *  省份(province)
 *  城市(city)
 *  网络方式(network)
 *  来源方式(source)
 *  浏览器类型(browserType)
 *  进入网站时间(entryTime)
 *  离开网站时间(leaveTime)
 *  用户的ID(userID)

1.  在`ClickLog`伴生对象中实现`apply`方法
2.  使用FastJSON的`JSON.parseObject`方法将JSON字符串构建一个`ClickLog`实例对象
3.  使用map算子将数据封装到`ClickLog`样例类
4.  在样例类中编写一个main方法，传入一些JSON字符串测试是否能够正确解析
5.  重新运行Flink程序，测试数据是否能够完成封装

使用JSONObject.get("key")获取到的数据是一个Any类型，要调用toString方法转换为String类型

**参考代码**  
在bean目录下创建

import com.alibaba.fastjson.JSON
    
    /** * 使用ClickLog样例类来封装点击流日志 */
    // - 频道ID(channelID)
    // - 产品类别ID(categoryID)
    // - 产品ID(produceID)
    // - 国家(country)
    // - 省份(province)
    // - 城市(city)
    // - 网络方式(network)
    // - 来源方式(source)
    // - 浏览器类型(browserType)
    // - 进入网站时间(entryTime)
    // - 离开网站时间(leaveTime)
    // - 用户的ID(userID)
    case class ClickLog (
                          var channelID:String,
                          var categoryID:String,
                          var produceID:String,
                          var country:String,
                          var province:String,
                          var city:String,
                          var network:String,
                          var source:String,
                          var browserType:String,
                          var entryTime:String,
                          var leaveTime:String,
                          var userID:String)
    object ClickLog { 
      def apply(json:String): ClickLog = { 
        val jsonObject = JSON.parseObject(json)
    
        val channelID = jsonObject.getString("channelID")
        val categoryID = jsonObject.getString("categoryID")
        val produceID = jsonObject.getString("produceID")
        val country = jsonObject.getString("country")
        val province = jsonObject.getString("province")
        val city = jsonObject.getString("city")
        val network = jsonObject.getString("network")
        val source = jsonObject.getString("source")
        val browserType = jsonObject.getString("browserType")
        val entryTime = jsonObject.getString("entryTime")
        val leaveTime = jsonObject.getString("leaveTime")
        val userID = jsonObject.getString("userID")
    
        ClickLog(
          channelID,
          categoryID,
          produceID,
          country,
          province,
          city,
          network,
          source,
          browserType,
          entryTime,
          leaveTime,
          userID
        )
      }
    
      def main(args: Array[String]): Unit = { 
          val clickLog = ClickLog("{\"browserType\":\"火狐\",\"categoryID\":20,\"channelID\":19,\"city\":\"ZhengZhou\",\"country\":\"china\",\"entryTime\":1544605260000,\"leaveTime\":1544634060000,\"network\":\"电信\",\"produceID\":7,\"province\":\"HeBeijing\",\"source\":\"百度跳转\",\"userID\":6}")
          println(clickLog)
        }
    }

修改一下 (ClickLog(message), count, timeStamp)

//JSON->元组
        val tupleDataStream = kafkaDataStream.map { 
          msgJSON => { 
            val jsonObject = JSON.parseObject(msgJSON)
    
            val message = jsonObject.getString("message")
            val count = jsonObject.getLong("count")
            val timeStamp = jsonObject.getLong("timeStamp")
    
            (ClickLog(message), count, timeStamp)
          }
        }
        tupleDataStream.print()

### 封装Kafka消息为Message样例类 ###

**步骤**

1.  创建一个`Message`样例类，将ClickLog、时间戳、数量封装
2.  将Kafka中的数据整个封装到Message类中
3.  运行Flink测试  
    **参考代码**  
    Message.scala

/** * 封装Kafka中的消息 * @param clickLog 点击流浏览数据Bean对象 * @param count 数量 * @param timestamp 时间戳 */
    case class Message(var clickLog:ClickLog,
                       var count:Long,
                       var timestamp:Long)

App.scala

object App { 
      def main(args: Array[String]): Unit = { 
    	...
        // 使用map算子，将kafka中消费到的数据
        val tupleDataStream = kafkaDataStream.map { 
          msgJSON => { 
            val jsonObject = JSON.parseObject(msgJSON)
    
            val message = jsonObject.getString("message")
            val count = jsonObject.getLong("count")
            val timeStamp = jsonObject.getLong("timeStamp")
    
         // (ClickLog(message), count, timeStamp)
            Message(ClickLog(message),count,timeStamp)
        }
        }
        ...
      }
    }

## Flink添加水印支持 ##

水印（watermark）就是一个`时间戳`，Flink可以给数据流添加水印，可以理解为：Flink收到一条消息后，额外给这个消息添加了一个时间字段，这就是`添加水印`。

*  水印并不会影响原有Eventtime
 *  一般会设置水印时间，比Eventtime小几秒钟
 *  当数据流添加水印后，会按照水印时间来触发窗口计算
 *  当接收到的`水印时间 >= 窗口的endTime`，则触发计算  
    **观察下图:**  
    `60号`消息的EventTime为`10:10:00`, 正好符合10:00:00到10:10:00这个时间窗口的endtime。正常情况, 该消息应该被这个窗口计算的。但是当发生网络延迟的时候，该消息可能会晚到几秒钟，那么当它到达flink时，该窗口已经运算完毕。为了解决该问题，我们为该消息设置`watermark时间`为`10:09:57`，当它到达flink时，会把该watermark时间设置为窗口的当前时间，由于小于endtime，所以该消息到达后并不会立即计算。直到一个携带watermark时间大于或者等于endtime的时候，窗口计算才会被触发。这样就可以有效的解决由于网络延迟造成的数据计算不精确的情况。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 1]  
    1.在`App.scala`中添加水印支持

tupleDataStream.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[Message]{ 
        // 当前时间戳
        var currentTimestamp = 0l
        //延迟时间
        var maxDelayTime = 2000l
    
        // 获取水印时间
        override def getCurrentWatermark: Watermark = { 
            new Watermark(currentTimestamp - maxDelayTime)
        }
    
        // 获取EventTime
        override def extractTimestamp(element: Message, previousElementTimestamp: Long): Long = { 
    
            // 比较当前消息的时间戳和上一个消息的时间戳,取最大值
            currentTimestamp = Math.max(element.timeStamp,previousElementTimestamp)
            currentTimestamp
        }
    })

2.启动执行测试，观察输出

Message(1,1557475915909,ClickLog(火狐,9,14,ShiJiaZhuang,china,1544612460000,1544634060000,移动,2,HeNan,必应跳转,11))

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]: /images/20221023/d4029b311da64a3ba830a975d02ea54e.png
[20210307150410957.png]: /images/20221023/c8094d08e2834d0e93a927c548fb3695.png
[20210307150520284.png]: /images/20221023/91b86e918d044ce591ce40b52a5d4475.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70 1]: /images/20221023/1888d8b0231a4432a77e163663d31636.png