在flink中设置事件时间时需要将时间的表示转换为毫秒
如果不需要转换
def main(args: Array[String]): Unit = {
// ...
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) // 将时间特性设置为事件时间
env.setParallelism(1)
val clickStream = env
.fromElements(
UserClickLog("user_2", "1500", "click", "page_1"),
UserClickLog("user_2", "2000", "click", "page_1")
)
.assignAscendingTimestamps(_.eventTime.toLong * 1000L) // 选择事件时间的字段
// ...
}如果需要转换
def main(args: Array[String]): Unit = {
// ...
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) // 将时间特性设置为事件时间
val clickStream = env
.fromElements(
UserClickLog("user_2", "2019-11-16 17:30:00", "click", "page_1")
)
.assignTimestampsAndWatermarks(
new BoundedOutOfOrdernessTimestampExtractor[UserClickLog](Time.seconds(0)) {
override def extractTimestamp(t: UserClickLog): Long = {
val dateTimeFormatter = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss")
val dateTime = DateTime.parse(t.eventTime, dateTimeFormatter)
dateTime.getMillis // 返回事件时间
}
}
)
// ...
}Time.seconds(0): MaxOutOfOrderness 延迟时间, 水位线用于延迟窗口的触发时间
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 [email protected] 举报，一经查实，本站将立刻删除。
