小编Xia*_*REN的帖子

我想知道是否可以通过使用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 ),

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”为数据流中的记录分配挂钟时间戳，是否有可能在运行时获得这样的时间戳？例如，这是我想要做的：

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

谢谢你的帮助。