一、map
map算子把input类型数据转换为output类型数据
eg：String input = “sensor,123456,33.0”
SensorReading(“sensor”,123456L,33.0d)
dataStream.map(new Mapfunction<input,output>{xxx})
import com.tan.flink.bean.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class Transform_Map {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> inputDataStream = env.socketTextStream("192.168.200.102", 9999);
SingleOutputStreamOperator<SensorReading> resultDataStream = inputDataStream.map(new MapFunction<String, SensorReading>() {
@Override
public SensorReading map(String input) throws Exception {
String[] fields = input.split(",");
String id = fields[0];
Long timestamp = Long.parseLong(fields[1]);
Double temperatur = Double.parseDouble(fields[2]);
return new SensorReading(id, timestamp, temperatur);
}
});
resultDataStream.print();
env.execute();
}
}
二、flatMap
flatMap 扁平化算子：把输入input类型转化为output类型输出，与map不同，flatmap输出多个output类型。
eg：String input = “hello,word”
output：Tuple2(“hello”,1)、Tuple2(“word”,1)
dataStream.flatMap(new FlatMapFunction<input,outpu>{xxx})
inputDataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
@Override
public void flatMap(String input, Collector<Tuple2<String, Integer>> collector) throws Exception {
String[] words = input.split(",");
for (String word : words) {
collector.collect(new Tuple2<>(word, 1));
}
}
});
三、filter
filter算子把input类型输入数据过滤出来 。为true留下来，false过滤掉。
eg：String input = “hello,world”
dataStream.filter(new FilterFunction{return boolean})
inputDataStream.filter(new FilterFunction<String>() {
@Override
public boolean filter(String input) throws Exception {
return input.contains("hello");
}
});
四、keyby
DataStream → KeyedStream：逻辑地将一个流拆分成不相交的分区，每个分区包含具有相同 key 的元素，在内部以 hash 的形式实现的。
eg：dataStream.keyby(param)
param 可以是输入数据字段下标，默认从0开始，比如Tuple2(word,1)，param=0，就是按照word分流。
param 可以是输入数据的字段名称，比如一个类Sensor有三个字段：id，timestamp,temperature。param=“id”，就是按照id进行分流。
还有其他KeySelector。后面结合滚动聚合算子进行案例。
五、滚动聚合算子（rolling Aggregation）
5.1、sum
案例：统计wordcount
SingleOutputStreamOperator<Tuple2<String, Integer>> wordDataStream = inputDataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
@Override
public void flatMap(String input, Collector<Tuple2<String, Integer>> collector) throws Exception {
String[] words = input.split(" ");
for (String word : words) {
collector.collect(new Tuple2<>(word, 1));
}
}
});
SingleOutputStreamOperator<Tuple2<String, Integer>> resultDataStream = wordDataStream.keyBy(0)
.sum(1);
5.2、max
选择每条流的最大值
案例：根据输入数据（比如 第一次输入 hello,9 、第二次输入 world,7，结果还是 hello，9）
inputDataStream.map(new MapFunction<String, Tuple2<String, Long>>() {
@Override
public Tuple2<String, Long> map(String input) throws Exception {
String[] fields = input.split(" ");
return new Tuple2(fields[0], Long.parseLong(fields[1]));
}
}).keyBy(0)
.max(1);
5.3、min
选择每条流的最小值
案例：根据输入数据（比如 第一次输入 hello,9 、第二次输入 world,7，结果是 hello，7）
inputDataStream.map(new MapFunction<String, Tuple2<String, Long>>() {
@Override
public Tuple2<String, Long> map(String input) throws Exception {
String[] fields = input.split(" ");
return new Tuple2(fields[0], Long.parseLong(fields[1]));
}
}).keyBy(0)
.min(1);
5.4、minby
针对keyedStream中的某个字段数据进行选择最小值
案例：根据每次传感器id传来的数据选择温度最小的
DataStreamSource<SensorReading> inputDataStream = env.addSource(new SourceFromCustom.CustomSource());
SingleOutputStreamOperator<SensorReading> resultDataStream = inputDataStream.keyBy("id")
.minBy("temperature");
5.5、maxby
原理同maxby
5.6、reduce
KeyedStream → DataStream：一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。
案例：根据传感器id传来的数据比较上一次时间戳的温度，选择最大温度的时间戳数据
import com.tan.flink.bean.SensorReading;
import com.tan.flink.source.SourceFromCustom;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class Transform_keyed_Reduce {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<SensorReading> inputDataStream = env.addSource(new SourceFromCustom.CustomSource());
SingleOutputStreamOperator<SensorReading> resultDataStream = inputDataStream.keyBy("id")
.reduce(new CustomReduceFunction());
resultDataStream.print();
env.execute();
}
public static class CustomReduceFunction implements ReduceFunction<SensorReading> {
@Override
public SensorReading reduce(SensorReading sensorReading, SensorReading input) throws Exception {
String id = sensorReading.getId();
Long timestamp = input.getTimestamp();
double temperature = Math.max(sensorReading.getTemperature(), input.getTemperature());
return new SensorReading(id, timestamp, temperature);
}
}
}
六、split和select
6.1、split
DataStream → SplitStream：根据某些特征把一个 DataStream 拆分成两个或者多个 DataStream。
6.2、select
SplitStream→DataStream：从一个 SplitStream 中获取一个或者多个
DataStream。
6.3、案例
根据传感器的温度，以60度为标准，大于等于60度为high流，其他为low流
import com.tan.flink.bean.SensorReading;
import com.tan.flink.source.SourceFromCustom;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class Transform_Split_Select {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<SensorReading> inputDatStream = env.addSource(new SourceFromCustom.CustomSource());
SplitStream<SensorReading> splitStream = inputDatStream.split(new OutputSelector<SensorReading>() {
@Override
public Iterable<String> select(SensorReading sensorReading) {
Double temperature = sensorReading.getTemperature();
if (temperature >= 60) {
return Lists.newArrayList("high");
} else {
return Lists.newArrayList("low");
}
}
});
DataStream<SensorReading> high = splitStream.select("high");
DataStream<SensorReading> low = splitStream.select("low");
DataStream<SensorReading> all = splitStream.select("high", "low");
high.print("high").setParallelism(1);
low.print("low").setParallelism(1);
all.print("all").setParallelism(1);
env.execute();
}
}
七、connect和CoMap
7.1、connect
DataStream,DataStream → ConnectedStreams：连接两个保持他们类型的数据流，两个数据流被 Connect 之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。输入数据类型可以一样也可以不一样。
7.2、CoMap、CoFlatMap
ConnectedStreams → DataStream：作用于 ConnectedStreams 上，功能与 map和 flatMap 一样，对 ConnectedStreams 中的每一个 Stream 分别进行 map 和 flatMap处理。最终处理的结果可以一样也可以不一样。
7.3、案例
根据传感器传来的温度数据分为高温度和低温度两条流进行connect，进行CoMap或者CoFlatMap算子之后输出数据。
import com.tan.flink.bean.SensorReading;
import com.tan.flink.source.SourceFromCustom;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
public class Transform_Connect_CoMap {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<SensorReading> inputDatStream = env.addSource(new SourceFromCustom.CustomSource());
SplitStream<SensorReading> splitStream = inputDatStream.split(new OutputSelector<SensorReading>() {
@Override
public Iterable<String> select(SensorReading sensorReading) {
Double temperature = sensorReading.getTemperature();
if (temperature >= 60) {
return Lists.newArrayList("high");
} else {
return Lists.newArrayList("low");
}
}
});
DataStream<SensorReading> high = splitStream.select("high");
SingleOutputStreamOperator<Tuple2<String, Double>> highDataStream = high.map(new MapFunction<SensorReading, Tuple2<String, Double>>() {
@Override
public Tuple2<String, Double> map(SensorReading sensorReading) throws Exception {
return new Tuple2<>(sensorReading.getId(), sensorReading.getTemperature());
}
});
DataStream<SensorReading> lowDataStream = splitStream.select("low");
ConnectedStreams<Tuple2<String, Double>, SensorReading> connectDataStream = highDataStream.connect(lowDataStream);
SingleOutputStreamOperator<Object> resultDataStream = connectDataStream.map(new CoMapFunction<Tuple2<String, Double>, SensorReading, Object>() {
@Override
public Object map1(Tuple2<String, Double> input) throws Exception {
// 处理高温数据
return new Tuple3<>(input.f0, input.f1, "warnning");
}
@Override
public Object map2(SensorReading input) throws Exception {
// 处理正常温度数据
return new Tuple3<>(input.getId(), input.getTimestamp(), input.getTemperature());
}
});
resultDataStream.print();
env.execute();
}
}
八、Union
DataStream → DataStream：对两个或者两个以上的DataStream进行union操作，产生一个包含所有DataStream元素的新DataStream。注意:如果你将一个DataStream跟它自己做union操作，在新的DataStream中，你将看到每一个元素都出现两次。
8.2、Connect 与 Union 区别：
Union 之前两个流的类型必须是一样，Connect 可以不一样，在之后的 coMap中可以再去调整成为一样的，也可以不一样的。
Connect 只能操作两个流，Union 可以操作多个。
8.1 案例
根据传感器传来的温度数据分为高温度和低温度两条流进行connect，进行union。
import com.tan.flink.bean.SensorReading;
import com.tan.flink.source.SourceFromCustom;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class Transform_Union {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<SensorReading> inputDatStream = env.addSource(new SourceFromCustom.CustomSource());
SplitStream<SensorReading> splitStream = inputDatStream.split(new OutputSelector<SensorReading>() {
@Override
public Iterable<String> select(SensorReading sensorReading) {
Double temperature = sensorReading.getTemperature();
if (temperature >= 60) {
return Lists.newArrayList("high");
} else {
return Lists.newArrayList("low");
}
}
});
DataStream<SensorReading> high = splitStream.select("high");
DataStream<SensorReading> low = splitStream.select("low");
DataStream<SensorReading> unionDataStream = high.union(low);
SingleOutputStreamOperator<Tuple3<String, Long, Object>> resultDataStream = unionDataStream.map(new MapFunction<SensorReading, Tuple3<String, Long, Object>>() {
@Override
public Tuple3<String, Long, Object> map(SensorReading input) throws Exception {
if (input.getTemperature() >= 60) {
return new Tuple3<String, Long, Object>(input.getId(), input.getTimestamp(), "warnning");
} else {
return new Tuple3<String, Long, Object>(input.getId(), input.getTimestamp(), input.getTemperature());
}
}
});
resultDataStream.print();
env.execute();
}
}