flink笔记10 [实验]体验ProcessingTime和指定EventTime下的区别

Posted 2021-05-29 Aurora1217

体验ProcessingTime和指定EventTime下的区别

 

实验数据

实验代码

实验结果

实验分析

---

实验数据

sensor_1,1619492107,36.2
sensor_1,1619492108,36.0
sensor_1,1619492109,36.5
sensor_1,1619492110,34.3
sensor_1,1619492111,34.3
sensor_1,1619492112,34.3
sensor_1,1619492113,34.3
sensor_1,1619492114,34.3
sensor_1,1619492115,34.3
sensor_1,1619492116,34.3
sensor_1,1619492117,34.3
sensor_1,1619492118,34.3
sensor_1,1619492119,34.3
sensor_1,1619492120,36.2
sensor_1,1619492121,36.1
sensor_1,1619492122,36.5
sensor_1,1619492123,35.2
sensor_1,1619492124,36.2
sensor_1,1619492125,36.1
sensor_1,1619492126,36.5
sensor_1,1619492127,36.2
sensor_1,1619492128,36.1
sensor_1,1619492129,36.3
sensor_1,1619492130,36.2
sensor_1,1619492131,36.2
sensor_1,1619492132,36.2
sensor_1,1619492170,36.2
sensor_1,1619492171,36.5
sensor_1,1619492172,37.3
sensor_1,1619492173,36.2
sensor_1,1619492174,36.3
sensor_1,1619492175,36.1
sensor_1,1619492176,36.6
sensor_1,1619492177,36.5
sensor_1,1619492178,36.1
sensor_1,1619492179,36.8
sensor_1,1619492180,36.1
sensor_1,1619492209,36.5

实验代码

(一) EventTime

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.streaming.api.windowing.time.Time
import org.tysy.window_api.sensorReading

import java.util.Date

object window_eventtime {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val inputStream = env.socketTextStream("localhost",7777)

    val resultStream = inputStream
      .map(data =>{
        val arr = data.split(",")
        sensorReading(arr(0),arr(1).toLong,arr(2).toDouble)
      })
      .assignAscendingTimestamps(_.timestamp * 1000)
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[sensorReading](Time.seconds(3)) {
        override def extractTimestamp(element: sensorReading): Long = element.timestamp * 1000L
      } )
      .keyBy(_.id)
      .timeWindow(Time.seconds(15))
      .reduce((data1,data2)=>sensorReading(data1.id,data2.timestamp,data1.temperature.min(data2.temperature)))
      .map(data =>(data,new Date().getTime))

    resultStream.print()

    env.execute("window_eventtime test")
  }
}

(二) ProcessingTime

import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.streaming.api.windowing.time.Time
import org.tysy.window_api.sensorReading

import java.util.Date

object window_processingtime {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputStream = env.socketTextStream("localhost",7777)

    val resultStream = inputStream
      .map(data =>{
        val arr = data.split(",")
        sensorReading(arr(0),arr(1).toLong,arr(2).toDouble)
      })
      .keyBy(_.id)
      .timeWindow(Time.seconds(15))
      .reduce((data1,data2)=>sensorReading(data1.id,data2.timestamp,data1.temperature.min(data2.temperature)))
      .map(data =>(data,new Date().getTime))

    resultStream.print()

    env.execute("window_processingtime test")
  }
}

实验结果

实验分析

①数据的最后一列是系统的时间戳。
②当使用的是ProcessingTime时，在同一个window里有多少条数据，和测试代码时在该window的时间内输入的代码条数有关，最后一列的时间戳可看出窗口大小15秒。
③当使用的是EventTime时，和输入数据的速度没有关系，只和实验中指定的时间戳大小有关。