Flink Time

awwewwbbb

发布于 2022-05-15 09:10:30

2930

发布于 2022-05-15 09:10:30

文章被收录于专栏：chaplinthink的专栏

基础概念

支持两种时间概念：

Processing Time 时间递增
Event Time 支持一定程度的乱序上一个 checkpoint 或者 savepoint 进行重放，是不是希望结果完全相同。如果希望结果完全相同，就只能用 Event Time；如果接受结果不同，则可以用 Processing Time。

watermark

一个watermark 代表了 watermark所包含的timestamp 数值,表示后来的数据已经再也没有小于或等于这个时间的了.

Flink 支持两种 watermark 生成方式:

在SourceFunction中产生

collectWithTimestamp 方法发送一条数据第一个参数就是我们要发送的数据第二个参数就是这个数据所对应的时间戳 emitWatermark 去产生一条 watermark：表示接下来不会再有时间戳小于等于这个数值记录

在使用DataStream API 的时候指定

DataStream.assignTimestampsAndWatermarks

建议生成的工作越靠近 DataSource 越好。这样会方便让程序逻辑里面更多的 operator 去判断某些数据是否乱序。

code demo:

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

        val dataStream: DataStream[Order] = env.socketTextStream("localhost", 9999).map(item => {
            val itemArray = item.split(",")
            Order(itemArray(0).toLong, itemArray(1), itemArray(2).toDouble)
        })

        val outputStream: DataStream[Order] = dataStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[Order](Time.seconds(0)) {
            override def extractTimestamp(element: Order): Long = element.timestamp * 1000L
        }).keyBy("category").timeWindow(Time.seconds(5)).apply(new MyWindowFunction)

        dataStream.print("Data")
        outputStream.print("Result")

        env.execute()
    }
}

class MyWindowFunction extends WindowFunction[Order, Order, Tuple, TimeWindow] {
    override def apply(key: Tuple, window: TimeWindow, input: Iterable[Order], out: Collector[Order]): Unit = {
        val timestamp = window.maxTimestamp()
        var sum: Double = 0
        for (elem <- input) {
            sum += elem.price
        }

        val category = key.asInstanceOf[Tuple1[String]].f0
        out.collect(Order(timestamp, category, sum))
    }
}