Apache Flink:如何在摄取时间模式下获取事件的时间戳? [英] Apache Flink: How to get timestamp of events in ingestion time mode?

问题描述

我想知道是否可以通过使用Flink的摄取时间模式来获取记录的时间戳.考虑以下 flink 代码示例(https://github.com/apache/flink/blob/master/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/join/WindowJoinSampleData.scala),

I am wondering is it possible to obtain the timestamp of record by using Flink's ingestion time mode. Considering the following flink code example (https://github.com/apache/flink/blob/master/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/join/WindowJoinSampleData.scala),

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)

val grades = WindowJoinSampleData.getGradeSource(env, rate)
val salaries = WindowJoinSampleData.getSalarySource(env, rate)

val joined = joinStreams(grades, salaries, windowSize)

...
case class Grade(name: String, level: Int) 
case class Salary(name: String, salary: Int)

默认情况下，等级和工资都不包含时间戳字段.但是，由于 Flink 允许使用ingestionTime"为数据流中的记录分配挂钟时间戳，是否有可能在运行时获得这样的时间戳?例如，这是我想要做的:

By default, neither grade nor salary contains timestamp field. However, since Flink allows to use "ingestionTime" to assign the wall clock timestamp to the records in data stream, is it possible to obtain such timestamp at runtime? For example, here is what I am trying to do:

val oldDatastream = env.addSource...  // Using ingestion time
val newDatastream = oldDatastream.map{record =>   
    val ts = getRecordTimestamp(record)
    // do some thing with ts
    }

感谢您的帮助.

推荐答案

Use ProcessFunction wich 给你一个 Context，你可以用它来获取元素的时间戳(无论是其摄取、处理或事件时间).

Use ProcessFunction wich gives you a Context, that you can use to get the element's timestamp (whether its ingestion, processing or event time).

这篇关于Apache Flink:如何在摄取时间模式下获取事件的时间戳?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！