前言
如上图，算子，也叫operation或transformation，是编写业务逻辑的核心单元，类似于Spark中的RDD。
本文会参考Flink Operators 通过实例的方式，以DataStream API来讲解，如有不对的地方，请给我留言。
注：流处理和批处理的一个重要区别是，流处理是“rolling”，也就是说数据会不断的源源流入，不像批处理，是一次性获取所有数据，然后再一起做处理。
一、Map
DataStream --> DataStream：输入一个参数产生一个参数，map的功能是对输入的参数进行转换操作。
Map算子是一进一出，如下图所示
代码：input：[0,1,2,3,4,5,6,7,8,9,10]，
output：[100,101,102,103,104,105,106,107,108,109,110]
public class TestMap {
public static void main(String[] args)throws Exception{
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream input = env.generateSequence(0,10);
DataStream plusOne = input.map(new MapFunction() {
@Override
public Longmap(Long value)throws Exception {
System.out.println("-----------" + value);
return value+100;
}
});
plusOne.print();
env.execute();
}
}
二、FlatMap
DataStream --> DataStream：输入一个参数，产生0、1或者多个输出，这个多用于拆分操作。
flatMap是一进多出，最常见的例子就是wordcount：
public static void main(String[] args)throws Exception{
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream input = env.fromElements(WORDS);
DataStream wordStream = input.flatMap(new FlatMapFunction() {
@Override
public void flatMap(String value, Collector out)throws Exception {
String[] tokens = value.toLowerCase().split("\\W+");
for(String word : tokens){
if(word.length()>0) {
out.collect(word);
}
}
}
});
wordStream.print();
env.execute("w");
}
三、Filter
DataStream --> DataStream：结算每个元素的布尔值，并返回为true的元素
public class TestFilter {
public static void main(String[] args)throws Exception{
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream input = env.generateSequence(-5,5);
input.filter(new FilterFunction() {
@Override
public boolean filter(Long value)throws Exception {
return value>0;
}
}).print();
//input.print();
env.execute();
}
}
四、KeyBy
DataSteam --> DataStream：逻辑地将一个流拆分成不相交的分区，每个分区包含具有相同key的元素，在内部以hash的形式实现的。以key来分组。
注意：以下类型无法作为key
1. POJO类，且没有实现hashCode函数
2. 任意形式的数组类型
key的创建参考：Keyed DataStream
KeyBy是根据key来进行分类，类似SQL中的groupBy，分类之后就可以求最大值、最小值、平均值、求和等
public class TestKeyBy {
public static void main(String[] args) throws Exception{
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Tuple4<String,String,String,Integer>> input = env.fromElements(TRANSCRIPT);
System.out.println("----------" + input.getParallelism());
KeyedStream<Tuple4<String,String,String,Integer>,Tuple> keyedStream = input.keyBy("f0");
keyedStream.maxBy("f3").print();
env.execute();
}
public static final Tuple4[] TRANSCRIPT = new Tuple4[]{
Tuple4.of("class1","张三","语文",100),
Tuple4.of("class1","李四","语文",78),
Tuple4.of("class1","王五","语文",99),
Tuple4.of("class2","赵六","语文",81),
Tuple4.of("class2","钱七","语文",59),
Tuple4.of("class2","马二","语文",97)
};
}
五、reduce
KeyedStream --> DataStream：滚动合并操作，合并当前元素和上一次合并的元素结果。
实例：
输出：将f3列的成绩叠加
六、fold
KeyedStream --> DataStream：用一个初始的一个值，与其每个元素进行滚动合并操作。相当于是一次折叠操作，这个算子在新的API中已经去除，比较鸡肋。
public class TestFold {
public static void main(String[] args)throws Exception{
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream> input = env.fromElements(TRANSCRIPT);
DataStream result = input.keyBy(0).fold("start", new FoldFunction, String>() {
@Override
public Stringfold(String accumulator, Tuple4 value)throws Exception {
return accumulator +"=" + value.f1;
}
});
result.print();
env.execute();
}
public static final Tuple4[]TRANSCRIPT =new Tuple4[]{
Tuple4.of("class1","张三","语文",100),
Tuple4.of("class1","李四","语文",78),
Tuple4.of("class1","王五","语文",99),
Tuple4.of("class2","赵六","语文",81),
Tuple4.of("class2","钱七","语文",59),
Tuple4.of("class2","马二","语文",97)
};
}
输出：
七、aggregation
KeyedStream --> DataStream：分组流数据的滚动聚合操作：min和minBy的区别是min返回的是一个最小值，而minBy返回的是其字段中包含的最小值的元素（同样元原理适用于max和maxBy）
八、iterate
DataStream --> IterativeStream --> DataStream：在流程中创建一个反馈循环，将一个操作的输出重定向到之前的操作，这对于定义持续更新模型的算法来说很有意义的。
九、aggregation on windows
WindowedStream --> DataStream：对window的元素做聚合操作，min和minBy的区别是min返回的是最小值，而minBy返回的是包含最小值字段的元素。（同样原理适用于max和maxBy）
十、connect 和union的区别
DataStream，DataStream --> ConnectedStreams：连接两个保持她们类型的数据流，各自分析，并且双流之间可以共享状态（比如计数），这在第一个流的输入会影响第二个流时，非常有用。
DataStream* --> DataStream：连接两个及以上相同的数据流，合并多个流，新的流包含所有输入的流。
注意：如果将一个DataStream和自己做union操作，在新的DataStream中，将看到每个元素重复两次
使用的算子是coMap、coFlatMap，类似于Map、FlatMap，只不过作用在ConnectedStreams。
public class TestConnect {
public static void main(String[] args)throws Exception{
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream somestream = env.generateSequence(0,10);
DataStream otherstream = env.fromElements(WORDS);
ConnectedStreams connectedStreams = somestream.connect(otherstream);
DataStream result = connectedStreams.flatMap(new CoFlatMapFunction() {
@Override
public void flatMap1(Long value, Collector out)throws Exception {
out.collect(value.toString());
}
@Override
public void flatMap2(String value, Collector out)throws Exception {
for(String word : value.split("\\W+")){
if(word.length()>0){
out.collect(word);
}
}
}
});
result.print();
env.execute();
}
public static final String[]WORDS =new String[]{
"To be, or not to be,--that is the question:--",
"Whether 'tis nobler in the mind to suffer",
"The slings and arrows of outrageous fortune",
"Or to take arms against a sea of troubles,",
"And by opposing end them?--To die,--to sleep,--",
"No more; and by a sleep to say we end",
"The heartache, and the thousand natural shocks",
"That flesh is heir to,--'tis a consummation",
"Devoutly to be wish'd. To die,--to sleep;--",
"To sleep! perchance to dream:--ay, there's the rub;",
"For in that sleep of death what dreams may come,",
"When we have shuffled off this mortal coil,",
"Must give us pause: there's the respect",
"That makes calamity of so long life;",
"For who would bear the whips and scorns of time,",
"The oppressor's wrong, the proud man's contumely,",
"The pangs of despis'd love, the law's delay,",
"The insolence of office, and the spurns",
"That patient merit of the unworthy takes,",
"When he himself might his quietus make",
"With a bare bodkin? who would these fardels bear,",
"To grunt and sweat under a weary life,",
"But that the dread of something after death,--",
"The undiscover'd country, from whose bourn",
"No traveller returns,--puzzles the will,",
"And makes us rather bear those ills we have",
"Than fly to others that we know not of?",
"Thus conscience does make cowards of us all;",
"And thus the native hue of resolution",
"Is sicklied o'er with the pale cast of thought;",
"And enterprises of great pith and moment,",
"With this regard, their currents turn awry,",
"And lose the name of action.--Soft you now!",
"The fair Ophelia!--Nymph, in thy orisons",
"Be all my sins remember'd."
};
}
输出：
第十一、split和select
split：DataStream --> SplitStream，按照指定标准将指定的DataStream拆分成多个SplitStream。
select：SplitStream --> DataStream，跟split搭配使用，从SplitStream中选择一个或多个流。
public class TestSplitAndSelect {
public static void main(String[] args)throws Exception{
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream input = env.generateSequence(0,10);
SplitStream splitStream = input.split(new OutputSelector() {
@Override
public Iterableselect(Long value) {
List output =new ArrayList();
if(value %2 ==0){
output.add("even");
}else {
output.add("odd");
}
return output;
}
});
//splitStream.print();
DataStream even = splitStream.select("even");
DataStream odd = splitStream.select("odd");
DataStream all = splitStream.select("even","odd");
even.print();
//odd.print();
//all.print();
env.execute();
}
}
打印偶数：
打印奇数：
打印全部（奇偶数）：
第十二、project
从Tuple中选择属性的子集，即仅限event数据类型为Tuple的DataStream
注意：只有java API
使用场景：ETL时删减计算过程中不需要的字段
public class TestProject {
public static void main(String[] args)throws Exception{
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream> input = env.fromElements(TRANSCRIPT);
DataStream> out = input.project(1,3);
out.print();
env.execute();
}
public static final Tuple4[]TRANSCRIPT =new Tuple4[]{
Tuple4.of("class1","张三","语文",100),
Tuple4.of("class1","李四","语文",78),
Tuple4.of("class1","王五","语文",99),
Tuple4.of("class2","赵六","语文",81),
Tuple4.of("class2","钱七","语文",59),
Tuple4.of("class2","马二","语文",97)
};
}
输出：
第十三、MapPartition
类似Map，一次仅处理一个分区的数据
并行度为2：
并行度为4：
第十四、Distinct，去重
返回数据中不相同的元素，可以指定去重所依据的字段
根据第一个元素去重：
不指定去重字段的话，就是全元素匹配：
第十五、SortPartition，分区内排序
分区和分组是两个不同的概念，不要混淆。
下面的例子是先在第一个字段升序，如果第一个字段相同，则根据第二个字段降序
输出：
第十六、Join（Default/Inner Join）
场景一：默认是等值连接，就是inner join
输出：
场景二：按照自定义类，格式化输出，如（用户id，用户名，城市名）
输出：
第十七、Outer Join
场景一、left outer join
输出：
场景二、right outer join
输出：
场景三、full outer join
输出：
第十八、笛卡尔积Cross
第十九、union
Flink源码：https://github.com/apache/flink
Flink官网：https://ci.apache.org/projects/flink/flink-docs-release-1.12/
