Flink之Window的使用（1）：計數窗口

相關文章鏈接

Flink之Window的使用（1）：計數窗口

Flink之Window的使用（2）：時間窗口

Flink之Window的使用（3）：WindowFunction的使用

具體實現代碼如下所示：

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")