Flink迭代流代码实战及原理分析【包含完整代码】

àì夳堔傛蜴生んèń 2023-02-15 13:45 23阅读 0赞

## 一.简介 ##

参考：[DataStream API编程指南][DataStream API]

## 二.代码实战 ##

package cn.streaming
    
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.api.java.utils.ParameterTool
    import org.apache.flink.streaming.api.scala.{ DataStream, StreamExecutionEnvironment}
    
    /** * 统计不及格、及格、良好、优秀的次数 <60 60 80 90 * Created by Administrator on 2020/6/5. */
    object DataStreamTest { 
        private final val Bound1 = 60
        private final val Bound2 = 80
        private final val Bound3 = 90
        def main(args: Array[String]): Unit = { 
          val params = ParameterTool.fromArgs(args)
          val env = StreamExecutionEnvironment.getExecutionEnvironment.setBufferTimeout(1)
          env.getConfig.setGlobalJobParameters(params)
    
          // 指定的IP和接口
          val hostname: String = "192.168.136.7"
          val port: Int = 9999
    
          // create input stream of integer pairs
          val inputStream : DataStream[(Int, Int)]  = env.socketTextStream(hostname, port, '\n')
            .flatMap(r => r.split("\\s"))
            .map(r => (r.toInt, 1))
    
          def compare(value: (Int,Int)) = value._1 < value._2 // 匿名函数
          val output  = inputStream
            .map(value => (value._1, Bound1, Bound2, Bound3, value._2))
            .iterate(
              (iteration: DataStream[(Int, Int, Int, Int, Int)]) => { 
                val feedback = iteration.filter(value => !compare(value._1, value._2))
                  .map(r => (r._1 - 1, r._2, r._3, r._4, r._5)) //继续向下执行
    
                Thread.sleep(100)
                feedback.print()
    
                val output = iteration.filter(value => compare(value._1, value._2)).map(r => (r._1, r._2, r._5)) // 不及格，直接输出
                (feedback, output)
              }
              , 5000L  // 5s计算一次
            )
    
          output.print()
          env.execute("Streaming Iteration Example")
        }
      }

## 三.入参及执行效果 ##

1.socket  
![在这里插入图片描述][20200605182904645.png]  
2.执行结果  
![在这里插入图片描述][20200605182924253.png]  
3.原理分析  
程序首先执行输出逻辑【也就是不需要迭代那部分】，打印出来两天分数小于60的数据，因Thread.sleep(100)，迭代部分会每100ms执行一次迭代，直至执行到不满足迭代条件为止。  
4.异常分析  
经验证，迭代只会在首次触发时执行，后续满足触发条件也不会触发！  
![在这里插入图片描述][20200605184013827.png]  
![在这里插入图片描述][20200605184033482.png]

[DataStream API]: https://blog.csdn.net/qq_27600723/article/details/106556774
[20200605182904645.png]: https://img-blog.csdnimg.cn/20200605182904645.png
[20200605182924253.png]: https://img-blog.csdnimg.cn/20200605182924253.png
[20200605184013827.png]: https://img-blog.csdnimg.cn/20200605184013827.png
[20200605184033482.png]: https://img-blog.csdnimg.cn/20200605184033482.png