1 前言
前面写了如何使用 Flink
读取常用的数据源,也简单介绍了如何进行自定义扩展数据源,本篇介绍它的下一步:数据转换 Transformation
,其中数据处理用到的函数,叫做算子 Operator
,下面是算子的官方介绍。
算子将一个或多个
DataStream
转换为新的DataStream
。程序可以将多种转换组合成复杂的数据流拓扑。
在学习过程中,官网是个不错的入门介绍,格式如下:
一共有两列,左边介绍了函数名称,转换前的流类型,以及转换后的流类型,右边进行了方法描述,介绍该算子的概念和作用,然后有个代码段,告诉了如何使用它。
但官网的中文介绍不多,有些例子 demo
也有点不完善,所以接下来我会将自己理解笔记贴下来,跟大家一起来学习。
友情提示,贴出来的代码段可能不完整,可以下载 github 上的代码进行参考。
2 转换 Transformation
2.1 Map,DataStream ---> DataStream
映射转换。输入是一个 DataStream
,输出也是一个 DataStream
,属于一对一操作,例如输入是 [1, 3, 5],然后每个数乘以 2,可以通过下面形式实现:
dataStream.map((MapFuction) <Integer, Integer>) value -> value * 2;);
将会输出 [2, 6, 10],也有可能不是如上顺序,默认按照程序处理数据的时间进行输出。
后面如果大家发现执行顺序与我展示的不一致,请不用担心,属于正常现象,由于顺序性问题涉及到时间和窗口属性的作用,所以请各位先按照默认情况运行,理解算子的概念和使用,之后再去了解上述概念。
2.2 FlatMap,DataStream ---> DataStream
"平坦"映射(不知道该如何翻译囧)。 不同于上面的单个操作,这是一对多操作,取一个元素并产生零个,一个或多个元素。例如拆分一行字符串,然后输出多个单词:
SingleOutputStreamOperator<String> operator = source.flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String value, Collector<String> out) throws Exception { String[] tokens = parseString2Tokens(value); if (tokens == null) { return; } for (String token : tokens) { out.collect(token); } } });
输出多个元素的关键是 Collector
这个参数,通过它可以收集到更多元素。
2.3 Filter,DataStream ---> DataStream
过滤操作。通过一个 boolean function
对元素进行过滤,保留为 true
的元素,从而达到过滤的目的。例如下面过滤操作,保留 id
是偶数的元素:
SingleOutputStreamOperator operator = source.filter((FilterFunction<String>) value -> { Student stu = parseTokens2Object(parseString2Tokens(value)); return stu != null && stu.getId() % 2 == 0; });
2.4 KeyBy,DataStream ---> KeyedStream
按键 key
进行分类。KeyBy
通过 Hash partitioning
方法将一个 stream
变成一组不相交的分区 partitions
,每个 partitions
包含的元素具有相同的 key
。例如输入的是一组字符串,根据第一个字段的整数值进行分类:
KeyedStream<String, Integer> keyedStream = source.keyBy((KeySelector<String, Integer>) value -> { String[] tokens = parseString2Tokens(value); return tokens == null ? 0 : Integer.valueOf(tokens[0]); });
注意,KeyBy
操作后,流类型从 DataStream
变成了 keyedStream
,是一组 partitions
。
上面介绍的是通用类型的 keySelector
分类方法,也可以通过下述两种方法进行分区:
keyedStream.keyBy(0); // key 是 tuple 中的第一个元素 keyedStream.keyBy("id"); // key 是 pojo 中的 id 字段,实现了 getter、setter
2.5 Reduce,KeyedStream ---> DataStream
对按 key
分类的数据流进行"滚动"压缩。可以将当前元素与前一个元素进行整合处理,并返回一个新值。
例如下面例子,通过 id
字段进行分类,然后使用 reduce
进行压缩处理,每次将学生的名字字段进行拼接,年龄进行相加,返回一个新的对象:
SingleOutputStreamOperator operator = source .map((MapFunction<String, Student>) value -> parseTokens2Object(parseString2Tokens(value))) .keyBy((KeySelector<Student, Integer>) value -> value == null ? 0 : value.getId()) .reduce((ReduceFunction<Student>) (value1, value2) -> { Student student = new Student(); student.setId(value1.getId() + value2.getId()); student.setName(value1.getName() + " || " + value2.getName()); student.setAge(value1.getAge() + value2.getAge()); return student; });
输出结果如下:
16> Student(id=2, name=name2, age=22) 11> Student(id=1, name=name1, age=21) 15> Student(id=3, name=name13, age=28) 16> Student(id=5, name=name10, age=25) 16> Student(id=10, name=name10 || name15, age=55) 16> Student(id=4, name=name2 || name7, age=44) 16> Student(id=6, name=name2 || name7 || name12, age=71) ... 16> Student(id=15, name=name10 || name15 || name5, age=80) 1> Student(id=12, name=name4 || name9 || name14, age=77) 11> Student(id=3, name=name1 || name6 || name11, age=68)
从结果可以看到,id
相同的都分到同一个分区(测试中可以简单通过前面的线程 id
确认,属于同一分区处理),然后传入新对象,按照 reduce
的操作进行了处理,返回了拼装之后的对象。
2.6 Fold,KeyedStream ---> DataStream
滚动折叠。合并当前元素和上一个被折叠的值,输入值可以与返回值类型不一样。
例如数据流是一组数字 [1, 5, 7],想要输出一个拼接后的字符串,可以通过下面进行处理:
// 标准格式 keyedStream.fold(${initialValue}, (s1, s2) -> s1 + " || " + s2); SingleOutputStreamOperator operator = source .map((MapFunction<String, Student>) value -> parseTokens2Object(parseString2Tokens(value))) .keyBy("id") .fold("strat", new FoldFunction<Student, String>() { @Override public String fold(String accumulator, Student value) throws Exception { return accumulator + " || " + value; } });
上述例子的 ${initialValue}
初始值是 start
,然后在方法中通过 ||
分隔符进行拼接,最后输出结果是:
16> strat || Student(id=2, name=name2, age=22) 11> strat || Student(id=1, name=name1, age=21) 15> strat || Student(id=3, name=name13, age=28) 16> strat || Student(id=2, name=name2, age=22) || Student(id=2, name=name12, age=27) 1> strat || Student(id=4, name=name9, age=24) 11> strat || Student(id=1, name=name1, age=21) || Student(id=1, name=name6, age=21) 1> strat || Student(id=4, name=name9, age=24) || Student(id=4, name=name14, age=29) 16> strat || Student(id=2, name=name2, age=22) || Student(id=2, name=name12, age=27) || Student(id=2, name=name7, age=22) ...
从输出结果可以看到,初始值和每次处理的对象进行了拼接,最后返回的是折叠后的对象,不过该方法被标注为 @Deprecated
,不建议继续使用。
2.7 Aggregations,KeyedStream ---> DataStream
在按 key
分类的数据流上滚动聚合。要注意的是 min
和 minBy
之间的区别是 min
返回指定字段的最小值,而 minBy
返回在此字段中具有最小值的元素,也可以理解成整个对象(max
和 maxBy
的机制相同)
常用的 sum
合计函数就不多说了,下面具体展示的 min
和 minBy
函数,取当前流中的最小值:
测试代码的逻辑是:从一个文本流中读取数据,转换成对象,接着根据 id
进行 keyBy
分类,接着从每组分区 partitions
中取出年龄 age
字段最小的,我们来分别测试 min
和 minBy
的不同之处:
2.7.1. 测试文件
# id, name, age, address (代替 index 下标) ... 1 name10 21 j 1 name9 20 k 1 name8 21 l 1 name7 21 m ... 1 name2 21 r 1 name1 10 s
为了便于从小样本中观察的具体的区别,id
字段都设成 1,它们将会分组到同一个 partitions
,然后 age
字段有三种值(21、20 和 10),每次处理一个元素时,都会输出一个最小值
2.7.2. 测试代码
SingleOutputStreamOperator operator = source .map((MapFunction<String, Student>) value -> { parseTokens2Object(parseString2Tokens(value)); } ) .keyBy("id") // 这里表示选择使用 minBy 或者 min 函数 .minBy("age")/min("age");
2.7.3. 查看结果
左图是 minBy
,右图是 min
输出结果,从结果可以看出,minBy
返回多个值的意思是:返回元素的完整对象,而 min
返回的是字段的最小值,其它字段将不会修改。于是左边能看到不同的 age
对应的下标 address
是不同的,而右边不同的 age
,对应的还是第一次输出的 address
结果。
综上所述,如果使用聚合函数 minBy
和 min
(以及 max
),注意自己需要返回的类型,是否需要一个完整的对象,如果是的话,请选择使用 minBy
。
2.8 Window,KeyedStream ---> WindowedStream
窗口。该函数允许在已分区的 KeyedStream
上定义窗口。例如最近 5 秒内到达的数据:
dataStream.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(5)));
2.9 WindowAll,DataStream ---> AllWindowedStream
在 DataStream
上定义窗口。与上面不同的是,此次定义的范围是所有流,windows
会根据某些特征(例如,最近 5 秒内到达的数据)对所有流事件进行划分窗口。
SingleOutputStreamOperator<Student> operator = source .map((MapFunction<String, Student>) value -> parseTokens2Object(parseString2Tokens(value))) .assignTimestampsAndWatermarks(new MyTimestampExtractor()) .windowAll(TumblingEventTimeWindows.of(Time.milliseconds(1))) .apply(new AllWindowFunction<Student, Student, TimeWindow>(){...};
注意:在许多情况下,这是非并行转换,所有记录都将被收集在 WindowAll
运算符的一项任务中。
2.10 Window Reduce、Apply、Aggregations,WindowedStream ---> DataStream
窗口函数跟前面提到的常用函数一样,也有属于窗口的数据流功能。
例如应用 apply
进行自定义逻辑加工,还有 reduce
进行“压缩”,返回合并后的值、sum
等统计函数。
唯一不同的是,前面常规函数对应的整个流中的数据,而 window
针对是根据特征(例如 5s 时间内)分开的窗口流的数据进行操作。
windowedStream.apply(new WindowFunction<...>{...}); windowedStream.sum(0; windowedStream.min("key"); windowedStream.maxBy(0);
2.11 Union,DataStream * ---> DataStream
联合。Union 将两个或多个数据流合并,生成一个新的数据流,包含了本身以及参数中数据流的所有数据。
dataStream.union(otherStream1, otherStream2);
如果是一个数据流联合它自己,相当于得到两份同样的数据。
2.12 Window Join, DataStream,DataStream ---> DataStream
窗口连接。通过 key
将同一个 window
的两个数据流 join
起来。
dataStream.join(otherStram) .where(<KeySelector>) .equalsTo(<KeySelector>) .window(TumblingEventTimeWindows.of(Time.seconds(3))) .apply(new JoinFunction<...>{...});
上面代码的作用是,在 3s 时间内连接两个窗口,连接条件是 where
(第一个流)和 equalsTo
(第二个流)相同的返回值。
2.13 Window coGroup,DataStream,DataStream ---> DataStream
窗口连接。跟上面的作用一样,通过 key
将同一个 window
的两个数据流连接起来。
dataStream.coGroup(otherStram) .where(<KeySelector>) .equalsTo(<KeySelector>) .window(TumblingEventTimeWindows.of(Time.seconds(3))) .apply(new CoGroupFunction<...>{...});
可以看到,除了 join
和 coGroup
函数不同外,还有一个 apply
处理使用的参数类型也不一样。
interface JoinFunction<IN1, IN2, OUT> { OUT join(IN1 first, IN2 second); } interface CoGroupFunction<IN1, IN2, O> { void coGroup(Iterable<IN1> first, Iterable<IN2> second, Collector<O> out); }
上面展示的是两个接口,在 join
算子中使用的是 JoinFunction
,对应的 join
方法是两个窗口流的单个入参;在 coGroup
算子中,使用的是 CoGroupFunction
,对应的 coGroup
方法是两个窗口流的迭代器,可以进行更多数据处理。
在参考资料四种详细介绍了两个的区别,如有连接操作,为了更多扩展性,推荐使用的是 CoGroup
算子。
2.14 Connect,DataStream,DataStream ---> ConnectedStreams
数据流连接。“连接”两个保存其类型的数据流,该连接允许两个流之间的共享状态。
例如下面连接了两个数据流,将会转换成 ConnectedStram
类型,接着使用 CoFlatMapFunction
分别对合并流中两种类型数据流进行处理(也可以自定义一个 MyCoFlatMapFunction
类,继承自 RichCoFlatMapFunction
),在里面实现自己的逻辑
ConnectedStreams<String, Student> connectedStreams = source.connect(studentDataStreamSource); connectedStreams.flatMap(new CoFlatMapFunction<String, Student, Object>() { @Override public void flatMap1(String value, Collector<Object> out) throws Exception { // 状态 1 out.collect("Add prefix : " + value); } @Override public void flatMap2(Student value, Collector<Object> out) throws Exception { // 状态 2 if (value.getId() % 2 != 0) { out.collect(value); } } });
前面的连接 Join
/CoGroup
数据流类型都必须是一致,而使用 Connecte
算子,允许两个数据流的类型不同,根据 flatMap1
(处理的是字符串) 和 flatMap2
(处理的是对象)中的 collecot
收集,进行输出:
7> Add prefix : 1 name1 21 a 7> Add prefix : 2 name2 22 b 6> Add prefix : 5 name15 30 o ... 6> Student(id=1, name=otherName1, age=1, address=, checkInTime=null, successTimeStamp=0) 8> Student(id=3, name=otherName3, age=3, address=, checkInTime=null, successTimeStamp=0)
2.15 Split,DataStream ---> SplitStream
切分。从 DataStream
转换成 SplitStream
,类似于字符串中的 split
切分方法,Split
算子作用就是根据设定的规则,将原来的流切分成多个流。例如根据 id
的奇偶数分成两个流:
SplitStream<String> splitStream = source.split(new OutputSelector<String>() { @Override public Iterable<String> select(String value) { String[] tokens = parseString2Tokens(value); int num = Integer.valueOf(tokens[0]); List<String> output = new ArrayList<>(); if (num % 2 == 0) { output.add("even"); } else { output.add("odd"); } return output; } });
2.16 Select,SplitStream ---> DataStream
选择。从拆分流中选择一个或多个流。接着上面的例子,如果需要从 SplitStream
中分别获取奇偶数的数据流,可以使用该算子:
DataStream<String> even = splitStream.select("even"); DataStream<String> odd = splitStream.select("odd"); DataStream<String> all = splitStream.select("even","odd");
在 Flink
1.9 版本中,切分流 SplitStream
也被标注为不建议使用 @Deprecated
,出现了更好的选择 Side Outputs,感兴趣的朋友可以深入看看~
2.17 Project,DataStream ---> DataStream
选择部分字段。注意,只对元组 Tuple 类型的输入流有效,输出的也是选择下标的新元组数据流。例如下面,选择是下标 1 和 3 的属性:
DataStreamSource<Tuple4<String, Integer, Long, BigDecimal>> customSource = env.fromCollection( Lists.newArrayList( Tuple4.of("one", 1, 1L, BigDecimal.ONE), Tuple4.of("two", 2, 2L, BigDecimal.ZERO), Tuple4.of("three", 3, 3L, BigDecimal.TEN), Tuple4.of("four", 4, 4L, BigDecimal.TEN) ) ); // 分离下标 1,3 到新到数据流 DataStream<Tuple2<Integer, BigDecimal>> tuple2DataStreamSource = customSource.project(1, 3);
输出的结果如下:
[1, 1] [2, 0] [3, 10] [4, 10]
3 总结
本篇主要介绍了不同算子的概念和作用,包括 map
/flatMap
/filter
/keyBy
/reduce
/fold
/aggregations
/window
/windowAll
/union
/join
/coGroup
/connect
/split
/select
/project
, 也指出它们转变前和转变后的数据流类型,其中重点介绍了 min
和 minBy
之间的区别,希望能够帮助大家在学习时更好理解每个算子的用法,在实际使用时,选择合适的。
测试算子功能和代码结构如下图:
将注释去掉就可以进行测试,调整不同的输入源和参数条件,来看看是否符合你的预期吧,如有困惑或不对之处,请与我交流~(可在 `Github` 中提 `Issue` 或掘金中联系)
4 项目地址
https://github.com/Vip-Augus/flink-learning-note
git clone https://github.com/Vip-Augus/flink-learning-note
5 参考资料
- Operators
- Flink 从 0 到 1 学习 —— Flink Data transformation(转换)
- Flink编程<一> 概念, Setup
- flink实战--双流join之Join和coGroup的区别和应用
- Side Outputs