Flink之Window的使用（1）：计数窗口

相关文章链接

具体实现代码如下所示：

val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)

val sensorStream: DataStream[SensorReading] = env
    .readTextFile("D:\Project\IDEA\bigdata-study\flink-demo\src\main\resources\source.txt")
    .map(new MyMapToSensorReading)

// 1、滚动窗口
val tumbleWindow: DataStream[SensorReading] = sensorStream
    .keyBy(_.id)
    .countWindow(5)
    .reduce((x, y) => SensorReading(x.id, y.timestamp, x.temperature + y.temperature))

// 2、滑动窗口
val slideWindows: DataStream[SensorReading] = sensorStream
    .keyBy(_.id)
    .countWindow(5, 2)
    .reduce((x, y) => SensorReading(x.id, y.timestamp, x.temperature + y.temperature))

slideWindows.print()

env.execute("CountWindowDemo")