1.使用 ReduceFunction函数
让两个元素结合起来，产生一个相同类型的元素，它是增量的，放在KeyBy函数之后
package flink.java.test;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class TestReduceFunctionOnWindow {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment() ;
DataStreamSource<Tuple3<String,String,Integer>> input = env.fromElements(ENGLISH);
input.keyBy(x -> x.f0)
.countWindow(2)
.reduce(new ReduceFunction<Tuple3<String, String, Integer>>() {
@Override
public Tuple3<String, String, Integer> reduce(Tuple3<String, String, Integer> value1, Tuple3<String, String, Integer> value2) throws Exception {
System.out.println("value1-->"+value1);
System.out.println("value2-->"+value2);
System.out.println("==========================");
return new Tuple3<>(value1.f0,value1.f1,value1.f2+value2.f2);
}
}).print("reduce累加");
env.execute() ;
}
public static final Tuple3[] ENGLISH = new Tuple3[]{
//班级 姓名 成绩
Tuple3.of("class1","张三",100),
Tuple3.of("class1","李四",30),
Tuple3.of("class1","王五",70),
Tuple3.of("class2","赵六",50),
Tuple3.of("class2","小七",40),
Tuple3.of("class2","小八",10),
};
}
执行返回结果：
value1-->(class2,赵六,50) value2-->(class2,小七,40) ========================== reduce累加:1> (class2,赵六,90) value1-->(class1,张三,100) value2-->(class1,李四,30) ========================== reduce累加:2> (class1,张三,130) Process finished with exit code 0
2.使用AggregateFunction函数统计计算
AggregateFunction 比 ReduceFunction 更加的通用，它有三个参数，一个输入类型（IN），一个累加器（ACC），一个输出类型（OUT）。
输入类型，就是输入流的类型。接口中有一个方法，可以把输入的元素和累加器累加。并且可以初始化一个累加器，然后把两个累加器合并成一个累加器，获得输出结果。
input.keyBy(x -> x.f0) .countWindow(2) // AggregateFunction 比 ReduceFunction 更加的通用，它有三个参数，一个输入类型（IN），一个累加器（ACC），一个输出类型（OUT） .aggregate(new AggregateFunction<Tuple3<String, String, Integer>, Tuple2<String,Integer>, Tuple2<String,Integer>>() {
// 创建累加器操作：初始化中间值 @Override public Tuple2<String, Integer> createAccumulator() { return Tuple2.of("class1",1000); }
// 累加器操作 @Override public Tuple2<String, Integer> add(Tuple3<String, String, Integer> value1, Tuple2<String, Integer> value2) { return Tuple2.of(value1.f0,value1.f2+value2.f1); }
// 获取结果 @Override public Tuple2<String, Integer> getResult(Tuple2<String, Integer> value) { return Tuple2.of(value.f0,value.f1); } // 累加器合并操作，只有会话窗口的时候才会调用！ @Override public Tuple2<String, Integer> merge(Tuple2<String, Integer> value, Tuple2<String, Integer> acc1) { return Tuple2.of(value.f0,value.f1+acc1.f1); } }) .print("aggregate累加") ; env.execute() ;
执行结果：
aggregate累加:1> (class2,1090) aggregate累加:2> (class1,1130) Process finished with exit code 0
3.ProcessWindowFunction(全窗口函数)
ProcessWindowFunction 有一个 Iterable 迭代器，用来获得窗口中所有的元素。
有一个上下文对象用来获得时间和状态信息，比其他的窗口函数有更大的灵活性。
但是这样做损耗了一部分性能和资源，因为元素不能增量聚合，相反 ，在触发窗口计算时，Flink 需要在内部缓存窗口的所有元素。
案例1：
input.keyBy(x -> x.f0)
.countWindow(2)
//public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window> extends AbstractRichFunction ...
.process(new ProcessWindowFunction<Tuple3<String, String, Integer>, Tuple3<String,String,Integer>, String, GlobalWindow>() {
@Override
public void process(String s, //参数1：key
Context context, //参数2：上下文对象
Iterable<Tuple3<String, String, Integer>> iterable, //参数3：这个窗口的所有元素
//参数4：收集器，用于向下游传递数据
Collector<Tuple3<String, String, Integer>> collector) throws Exception {
System.out.println(context.window().maxTimestamp());
int sum = 0 ;
String name = "" ;
for (Tuple3<String,String,Integer> tuple3:iterable){
sum += tuple3.f2 ;
name = tuple3.f1 ;
}
collector.collect(Tuple3.of(s,name,sum));
}
}).print();
输出结果：
9223372036854775807 2> (class1,李四,130) 9223372036854775807 1> (class2,小七,90) Process finished with exit code 0
案例2：
.process(new ProcessWindowFunction<Tuple2<String, Long>, Tuple2<String, Long>, String, TimeWindow>() {
// 参数1: key 参数2: 上下文对象 参数3: 这个窗口内所有的元素 参数4: 收集器, 用于向下游传递数据
@Override
public void process(String key,
Context context,
Iterable<Tuple2<String, Long>> elements,
Collector<Tuple2<String, Long>> out) throws Exception {
System.out.println(context.window().getStart());
long sum = 0L;
for (Tuple2<String, Long> t : elements) {
sum += t.f1;
}
out.collect(Tuple2.of(key, sum));
}
})
参考：