Flink之滑动窗口（SlidingEventTimeWindows）_flink滑动窗口 - CSDN文库

5星 · 超过95%的资源 4 浏览量更新于2023-03-16 评论收藏 18KB PDF 举报

身份认证购VIP最低享 7 折!

领优惠券(最高得80元）

资源详情

资源评论

资源推荐

Flink之滑动窗口（之滑动窗口（SlidingEventTimeWindows））

def main(args: Array[String]): Unit = {

// 环境

val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

env.setParallelism(1)

val dstream: DataStream[String] = env.socketTextStream("hadoop1",7777)

val textWithTsDstream: DataStream[(String, Long, Int)] = dstream.map { text =>

val arr: Array[String] = text.split(" ")

(arr(0), arr(1).toLong, 1)

}

val textWithEventTimeDstream: DataStream[(String, Long, Int)] =

textWithTsDstream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[(String, Long,

Int)](Time.milliseconds(1000)) {

override def extractTimestamp(element: (String, Long, Int)): Long = {

return element._2

}

})

val textKeyStream: KeyedStream[(String, Long, Int), Tuple] = textWithEventTimeDstream.keyBy(0)

textKeyStream.print("textkey:")

val windowStream: WindowedStream[(String, Long, Int), Tuple, TimeWindow] =

textKeyStream.window(SlidingEventTimeWindows.of(Time.seconds(2),Time.milliseconds(500)))

val groupDstream: DataStream[mutable.HashSet[Long]] = windowStream.fold(new mutable.HashSet[Long]()) { case (set,

(key, ts, count)) =>

set += ts

}

groupDstream.print("window::::").setParallelism(1)

env.execute()

}

作者：qq_37969476

本内容试读结束，登录后可阅读更多

下载后可阅读完整内容，剩余0页未读，立即下载

评论10

丛乐

2023-06-09

学习了如何设定滑动窗口大小和数据划分，能够更好地应对不同数据需求。

weixin_38701407

粉丝: 5
资源: 918

会员权益专享

图片转文字

全年可省5，000元立即开通

最新资源

资源上传下载、课程学习等过程中有任何疑问或建议，欢迎提出宝贵意见哦~我们会及时处理！点击此处反馈