提出缺点
ProcessWindowFunction 缺点
窗口先 缓存所有元素，等到触发时，对窗口内的所有元素执行计算，这种方式效率低。
AggregateFunction 缺点
来一条数据，就处理一条数据，效率更高，但无法计算平均值、中位数
解决
Flink 提供一个方法，可以把增量函数 ProcessWindowFunction 和全量函数 AggregateFunction 结合 使用
对于一个窗口来说，Flink 先增量计算，窗口关闭前，将增量计算结果发送给 ProcessWindowFunction 作为输入再进行处理。
案例
城市的气温数据，每隔30秒，获得该城市的这30秒内的温度最大值、最小值、平均值
准备数据
在工程目录下创建 data/qiwen.txt，内容如下：
江苏,苏州,20.5,2021-01-29 16:00:00
江苏,盐城,24.5,2021-01-29 16:00:20
江苏,盐城,28.5,2021-01-29 16:00:30
江苏,盐城,23.5,2021-01-29 16:00:40
江苏,苏州,21.5,2021-01-29 16:00:10
江苏,苏州,22.5,2021-01-29 16:00:20
江苏,苏州,24.5,2021-01-29 16:00:30
江苏,苏州,23.5,2021-01-29 16:00:40
江苏,苏州,22.5,2021-01-29 16:00:50
java
package std2;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
public class Std1 {
static class TempRecord {
private String province;
private String city;
private Double temp;
private Date time;
public TempRecord(String province, String city, Double temp, Date time) {
this.province = province;
this.city = city;
this.temp = temp;
this.time = time;
}
public String getProvince() {
return province;
}
public void setProvince(String province) {
this.province = province;
}
public String getCity() {
return city;
}
public void setCity(String city) {
this.city = city;
}
public Double getTemp() {
return temp;
}
public void setTemp(Double temp) {
this.temp = temp;
}
public Date getTime() {
return time;
}
public void setTime(Date time) {
this.time = time;
}
}
static class TempRecordAggsResult {
private String key;
private Date beginTime;
private Date endTime;
// flink运算时，要将较大的值赋给该属性，所以要初始化最小值
private Double max = Double.MIN_VALUE;
// flink运算时，要将较小的值赋给该属性，所以要初始化最大值
private Double min = Double.MAX_VALUE;
private Double sum = 0.0;
private Double avg = 0.0;
private Integer count = 0;
public TempRecordAggsResult() {
}
@Override
public String toString() {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String beginTimeStr = sdf.format(beginTime);
String endTimeStr = sdf.format(endTime);
return "TempRecordAggsResult{" +
"key='" + key + '\'' +
", beginTime=" + beginTimeStr +
", endTime=" + endTimeStr +
", max=" + max +
", min=" + min +
", sum=" + sum +
", avg=" + avg +
", count=" + count +
'}';
}
public String getKey() {
return key;
}
public void setKey(String key) {
this.key = key;
}
public Date getBeginTime() {
return beginTime;
}
public void setBeginTime(Date beginTime) {
this.beginTime = beginTime;
}
public Date getEndTime() {
return endTime;
}
public void setEndTime(Date endTime) {
this.endTime = endTime;
}
public Double getMax() {
return max;
}
public void setMax(Double max) {
this.max = max;
}
public Double getMin() {
return min;
}
public void setMin(Double min) {
this.min = min;
}
public Double getSum() {
return sum;
}
public void setSum(Double sum) {
this.sum = sum;
}
public Double getAvg() {
return avg;
}
public void setAvg(Double avg) {
this.avg = avg;
}
public Integer getCount() {
return count;
}
public void setCount(Integer count) {
this.count = count;
}
}
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<String> source = env.readTextFile("./data/qiwen.txt");
SingleOutputStreamOperator<TempRecord> dataStream = source
.flatMap(new FlatMapFunction<String, TempRecord>() {
@Override
public void flatMap(String value, Collector<TempRecord> out) throws Exception {
String[] arr = value.split(",");
Double temp = 0.0;
if (arr[2] != null) {
temp = Double.parseDouble(arr[2]);
}
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date time = sdf.parse(arr[3]);
TempRecord tempRecord = new TempRecord(arr[0], arr[1], temp, time);
out.collect(tempRecord);
}
})
.assignTimestampsAndWatermarks(
WatermarkStrategy.<TempRecord>forBoundedOutOfOrderness(Duration.ofSeconds(0))
.withTimestampAssigner(new SerializableTimestampAssigner<TempRecord>() {
@Override
public long extractTimestamp(TempRecord element, long recordTimestamp) {
return element.getTime().getTime();
}
})
);
SingleOutputStreamOperator<TempRecordAggsResult> result = dataStream
.keyBy(TempRecord::getCity)
.window(TumblingEventTimeWindows.of(Time.seconds(30)))
//增量计算
.aggregate(new AggregateFunction<TempRecord, TempRecordAggsResult, TempRecordAggsResult>() {
@Override
public TempRecordAggsResult createAccumulator() {
return new TempRecordAggsResult();
}
@Override
public TempRecordAggsResult add(TempRecord value, TempRecordAggsResult accumulator) {
Double maxTemp = value.getTemp() > accumulator.getMax() ? value.getTemp() : accumulator.getMax();
Double minTemp = value.getTemp() < accumulator.getMin() ? value.getTemp() : accumulator.getMin();
accumulator.setKey(value.getProvince() + "," + value.getCity());
accumulator.setMax(maxTemp);
accumulator.setMin(minTemp);
accumulator.setSum(value.getTemp() + accumulator.getSum());
accumulator.setCount(accumulator.getCount() + 1);
return accumulator;
}
@Override
public TempRecordAggsResult getResult(TempRecordAggsResult accumulator) {
return accumulator;
}
@Override
public TempRecordAggsResult merge(TempRecordAggsResult a, TempRecordAggsResult b) {
a.setMax(a.getMax() > b.getMax() ? a.getMax() : b.getMax());
a.setMin(a.getMin() < b.getMin() ? a.getMin() : b.getMin());
a.setSum(a.getSum() + b.getSum());
a.setCount(a.getCount() + b.getCount());
return a;
}
},
//全量计算
new ProcessWindowFunction<TempRecordAggsResult, TempRecordAggsResult, String, TimeWindow>() {
@Override
public void process(String s, ProcessWindowFunction<TempRecordAggsResult, TempRecordAggsResult, String, TimeWindow>.Context context, Iterable<TempRecordAggsResult> elements, Collector<TempRecordAggsResult> out) {
long windowStart = context.window().getStart();
long windowEnd = context.window().getEnd();
if (elements.iterator().hasNext()) {
TempRecordAggsResult result = elements.iterator().next();
// System.out.println("result:" + result.toString());
result.setBeginTime(new Date(windowStart));
result.setEndTime(new Date(windowEnd));
result.setAvg(result.getSum() / result.getCount());
out.collect(result);
}
}
});
result.print();
env.execute();
}
}
参考：
https://blog.csdn.net/winterking3/article/details/115352463