Flink 系例之 Fold

Fold算子：将数据流的每一次输出进行滚动叠加，合并输出结果

示例环境

java.version: 1.8.x
flink.version: 1.11.1

示例数据源（项目码云下载）

Flink 系例之搭建开发环境与数据

Fold.java

import com.flink.examples.DataSource;
import org.apache.flink.api.common.functions.FoldFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
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 java.util.List;/*** @Description Fold算子：将数据流的每一次输出进行滚动叠加，合并输出结果* （与Reduce的区别是，Reduce是拿前一次聚合结果累加后一次的并输出数据流；Fold是直接将当前数据对象追加到前一次叠加结果上并输出数据流）*/
public class Fold {/*** 遍历集合，分区打印每一次滚动叠加的结果（示例：按性别分区，按排序，未位追加输出）* @param args* @throws Exception*/public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(4);List<Tuple3<String,String,Integer>> tuple3List = DataSource.getTuple3ToList();//注意：使用Integer进行分区时，会导致分区结果不对，转换成String类型输出key即可正确输出KeyedStream<Tuple3<String,String,Integer>, String> keyedStream = env.fromCollection(tuple3List).keyBy(new KeySelector<Tuple3<String,String,Integer>, String>() {@Overridepublic String getKey(Tuple3<String, String, Integer> tuple3) throws Exception {//f1为性别字段,以相同f1值（性别）进行分区return String.valueOf(tuple3.f1);}});SingleOutputStreamOperator<String> result = keyedStream.fold("同学：", new FoldFunction<Tuple3<String, String, Integer>, String>() {@Overridepublic String fold(String s, Tuple3<String, String, Integer> tuple3) throws Exception {if (s.startsWith("男") || s.startsWith("女")){return s + tuple3.f0 + "、";} else {return (tuple3.f1.equals("man") ? "男" : "女") + s + tuple3.f0 + "、";}}});result.print();env.execute("flink Fold job");}
}

打印结果

2> 男同学：张三、
2> 男同学：张三、王五、
2> 男同学：张三、王五、吴八、
1> 女同学：李四、
1> 女同学：李四、刘六、
1> 女同学：李四、刘六、伍七、

Flink 系例之 Fold相关推荐

Flink 系例之 CountWindowAll
countWindowAll 数量窗口 (不分区数量滚动窗口[滑动窗口与滚动窗口的区别,在于滑动窗口会有数据元素重叠可能,而滚动窗口不存在元素重叠]) 示例环境 java.version: 1.8.x ...
Flink 系例之 Connectors 连接 Redis
通过使用 Flink DataStream Connectors 数据流连接器连接到 Redis 缓存数据库,并提供数据流输入与输出操作: 示例环境 java.version: 1.8.x flink ...
Flink 系例之 Connectors 连接 Kafka
通过使用 Flink DataStream Connectors 数据流连接器连接到 ElasticSearch 搜索引擎的文档数据库 Index,并提供数据流输入与输出操作: 示例环境 java.v ...
Flink 系例之 Connectors 连接 ElasticSearch
通过使用 Flink DataStream Connectors 数据流连接器连接到 ElasticSearch 搜索引擎的文档数据库 Index,并提供数据流输入与输出操作: 示例环境 java.v ...
Flink 系例之 Project
Project算子:从数据流的元数组中,重新排例参数并指定不同的下标位,返回新的数据流示例环境 java.version: 1.8.x flink.version: 1.11.1 示例数据源 (项目 ...
Flink 系例之 Connectors 连接 MySql
通过使用 Flink DataStream Connectors 数据流连接器连接到 Mysql 数据源,并基于 JDBC 提供数据流输入与输出操作: 示例环境 java.version: 1.8.x ...
Flink 系例之 DataStream Connectors 与示例模块
官方介绍 Flink 中的 API Flink 为流式 / 批式处理应用程序的开发提供了不同级别的抽象. Flink API 最底层的抽象为有状态实时流处理.其抽象实现是 Process Funct ...
java dataset flatmap_Flink 系例之 FlatMap
FlatMap算子:将数据流一行按逻辑或规则拆分成0行或多行输出示例环境 java.version: 1.8.x flink.version: 1.11.1 示例数据源 FlatMap.java i ...
vba 单元格一系例操作
1.单元格取值赋值 Private Sub CommandButton1_Click() Range("A1") = "111" Range("A2& ...
python 案例串接_Python基础系例--字典串操作
字符串字符串是由数字,字母.下划线组成的一串字符创建字符串,可以使用单引号和双引号: var1 ='Hello World!' var2="Hello World!" 访问字符 ...

Flink 系例之 Fold

Flink 系例之 Fold相关推荐

最新文章

热门文章

Flink 系例 之 Fold

Flink 系例 之 Fold相关推荐

最新文章

热门文章

Flink 系例之 Fold

Flink 系例之 Fold相关推荐