说明
ProcessWindowFunction 等到窗口 截止，再对 全部数据 进行统计
可以实现 中位数、平均值 这类，需要获取 全部数据才能统计的功能
比 windowFunction 提供了更多的上下文信息
调用方式
stream
.keyBy( ... )
.window( ... )
.process( ProcessWindowFunction 对象);
当窗口到达结束时间时，触发计算，就会调用 process() 方法
接口定义
public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window> extends AbstractRichFunction
public abstract void process(KEY key, Context context, Iterable<IN> elements, Collector<OUT> out) throws Exception;
}
方法 process 的形参解释：
KEY key：keyContext context：上下文，可获取windows开始时间、结束时间Iterable<IN> elements：当前窗口中接收的全部数据，是可迭代对象Collector<OUT> out：返回结果
keyBy() 后，即：数据分组后，每组执行一次 process() 方法
缺点
接收到所有数据后，再做处理，在 process 中迭代数据，这种方式效率较差。来一条数据，就处理一条数据，效率更高
案例
按照 年月 统计订单平均价
数据
2021-10-01,java从入门到入坟,80
2021-10-21,mysql从入门到入坟,50
2021-9-05,hadoop从入门到放弃,90
2021-9-15,zookeeper从入门到放弃,40
2021-9-25,flink从入门到放弃,30
思路
- 根据字符串，进行拆分，并封装到对象中
keyBy()根据年-月进行分组- 实现
process()，按照年月统计订单平均价
java
package top.malaoshi;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
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.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
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.util.Date;
import java.util.Iterator;
public class WindowProcess全量窗口统计订单平均价 {
/**
* 数据格式：
* 日期时间,商品名,价格
* 2021-10-01,java从入门到入坟,80
* 2021-10-21,mysql从入门到入坟,50
* 2021-9-05,hadoop从入门到放弃,90
* 2021-9-15,zookeeper从入门到放弃,40
* 2021-9-25,flink从入门到放弃,30
*
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断
DataStream<String> dataStream = env.socketTextStream("hadoop1", 9999);
SingleOutputStreamOperator<Goods> soso = dataStream.map(new MapFunction<String, Goods>() {
@Override
public Goods map(String s) throws Exception {
String[] arr = s.split(",");
Goods goods = new Goods();
goods.setDate(arr[0]);
goods.setTitle(arr[1]);
goods.setPrice(Double.parseDouble(arr[2]));
return goods;
}
});
KeyedStream<Goods, String> ks = soso.keyBy(new KeySelector<Goods, String>() {
/**
*
* @param goods：传入的元祖，格式：(广告id,点击数量)
* @return
* @throws Exception
*/
@Override
public String getKey(Goods goods) throws Exception {
String date = goods.getDate();
int index = date.lastIndexOf("-");
String yearMonth = date.substring(0, index);
return yearMonth;
}
});//流处理中的分组
//统计各个key出现的次数
SingleOutputStreamOperator<Result> agg = ks.window(TumblingProcessingTimeWindows.of(Time.seconds(15)))
.process(new ProcessWindowFunctionImpl());
agg.print();
//启动并等待程序结束，可传入任务名
env.execute();
}
static class ProcessWindowFunctionImpl extends ProcessWindowFunction<Goods, Result, String, TimeWindow> {
/**
*
* @param s key
* @param context 上下文，可获取窗口开始、结束时间
* @param iterable 当前窗口中接收的全部数据
* @param collector 返回结果
* @throws Exception
*/
@Override
public void process(String s, Context context, Iterable<Goods> iterable, Collector<Result> collector) throws Exception {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String windowStart = sdf.format(new Date(context.window().getStart()));
String windowEnd = sdf.format(new Date(context.window().getEnd()));
System.out.println(windowStart + "--------" + windowEnd+"---"+s);
Iterator<Goods> iterator = iterable.iterator();
int count = 0;
double totalPrice = 0.0;
while (iterator.hasNext()) {
Goods goods = iterator.next();
System.out.println("--" + goods.getTitle()+"--"+goods.getDate());
totalPrice += goods.getPrice();
count++;
}
if (count != 0) {
double avg = totalPrice / count;
Result result = new Result(s, avg);
collector.collect(result);
}
}
}
static class Goods {
String date;
String title;
Double price;
public String getTitle() {
return title;
}
public void setTitle(String title) {
this.title = title;
}
public String getDate() {
return date;
}
public void setDate(String date) {
this.date = date;
}
public Double getPrice() {
return price;
}
public void setPrice(Double price) {
this.price = price;
}
}
static class Result {
String date;
Double avg;
public Result(String date, Double avg) {
this.date = date;
this.avg = avg;
}
public String getDate() {
return date;
}
public void setDate(String date) {
this.date = date;
}
public Double getAvg() {
return avg;
}
public void setAvg(Double avg) {
this.avg = avg;
}
@Override
public String toString() {
return "Result{" +
"date='" + date + '\'' +
", avg=" + avg +
'}';
}
}
}
测试
在linux 执行下面命令：
nc -lk 9999
启动java程序
在 nc 窗口输入下面内容：
2021-10-01,java从入门到入坟,80
2021-10-21,mysql从入门到入坟,50
2021-9-05,hadoop从入门到放弃,90
2021-9-15,zookeeper从入门到放弃,40
2021-9-25,flink从入门到放弃,30
观察 idea 控制台，按照 年月 统计订单平均价
如下：
2021-12-14 10:48:00--------2021-12-14 10:48:15---2021-9
--zookeeper从入门到放弃--2021-9-15
2021-12-14 10:48:00--------2021-12-14 10:48:15---2021-10
--flink从入门到放弃--2021-9-25
--hadoop从入门到放弃--2021-9-05
--java从入门到入坟--2021-10-01
--mysql从入门到入坟--2021-10-21
2> Result{date='2021-10', avg=65.0}
4> Result{date='2021-9', avg=53.333333333333336}
每组执行一次 process() 方法