Flink是下一代大数据计算平台，可处理流计算和批量计算。《Flink-1.9流计算开发：七、fold函数》是cosmozhu写的本系列文章的第七篇。通过简单的DEMO来演示fold函数执行的效果 。
需求
本篇文章，我们使用fold函数每次来统一输出当前所有订单的分类汇总信息。
解决方案
代码语言：javascript复制public class StreamTest {
private static final Logger LOG = LoggerFactory.getLogger(StreamTest.class);
private static final String[] TYPE = { "苹果", "梨", "西瓜", "葡萄", "火龙果" };
@SuppressWarnings("deprecation")
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//添加自定义数据源,每秒发出一笔订单信息{商品名称,商品数量}
DataStreamSource<Tuple2<String, Integer>> orderSource = env.addSource(new SourceFunction<Tuple2<String, Integer>>() {
private volatile boolean isRunning = true;
private final Random random = new Random();
@Override
public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
while (isRunning) {
TimeUnit.SECONDS.sleep(1);
ctx.collect(Tuple2.of(TYPE[random.nextInt(TYPE.length)], 1));
}
}
@Override
public void cancel() {
isRunning = false;
}
}, "order-info");
//这里只为将DataStream → KeyedStream,用空字符串做分区键。所有数据为相同分区
orderSource.keyBy(new KeySelector<Tuple2<String,Integer>, String>(){
@Override
public String getKey(Tuple2<String, Integer> value) throws Exception {
return "";
}
})
//这里用HashMap做暂存器
.fold(new HashMap<String, Integer>(), new FoldFunction<Tuple2<String,Integer>, Map<String, Integer>>() {
@Override
public Map fold(Map<String, Integer> accumulator, Tuple2<String, Integer> value) throws Exception {
accumulator.put(value.f0, (Integer)accumulator.getOrDefault(value.f0, 0) value.f1);
return accumulator;
}
})
.print();
env.execute("Flink Streaming Java API Skeleton");
}
}执行效果
代码语言：javascript复制3> {苹果=1}
3> {苹果=1, 西瓜=1}
3> {苹果=1, 火龙果=1, 西瓜=1}
3> {苹果=2, 火龙果=1, 西瓜=1}
3> {苹果=2, 火龙果=1, 西瓜=2}
3> {苹果=2, 火龙果=1, 西瓜=3}
3> {苹果=2, 火龙果=1, 梨=1, 西瓜=3}
3> {苹果=3, 火龙果=1, 梨=1, 西瓜=3}
3> {苹果=3, 火龙果=1, 梨=1, 西瓜=4}
3> {苹果=3, 火龙果=1, 梨=1, 西瓜=5}
3> {苹果=3, 火龙果=1, 梨=2, 西瓜=5}
3> {苹果=3, 火龙果=1, 梨=2, 西瓜=6}
3> {苹果=3, 火龙果=2, 梨=2, 西瓜=6}小结
这块代码有几个注意的地方：
1、keyBy -> return ""。 fold 函数是 keyedStream 流的方法，为了将DataStream不做其它处理的转为 keyedStream。因此在这里使用了这种写法。
2、new HashMap<String, Integer>() 这里是创建一个临时暂存器，用来保存流中产生的状态。
代码地址
代码语言：javascript复制https://github.com/chaoxxx/learn-flink-stream-api/blob/master/src/main/java/fun/cosmozhu/session7/StreamTest.java作者：cosmozhu --90后的老父亲，专注于保护地球的程序员
个人网站：https://www.cosmozhu.fun
欢迎转载，转载时请注明出处。
相关文章
- Flink-1.9流计算开发：十六、intervalJoin函数
- Flink-1.9流计算开发：十五、join函数
- Flink-1.9流计算开发：十四、union函数
- Flink-1.9流计算开发：十三、min、minBy、max、maxBy函数
- Flink-1.9流计算开发：十二、apply函数
