1:Map映射

one-to-one 算子，将一种数据类型转换成另外一种数据类型

eg:

package cn.itcast.flink.transformation;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.text.SimpleDateFormat;/*** 使用map操作，读取apache.log文件中的字符串数据转换成ApacheLogEvent对象*/
public class MapDemo {public static void main(String[] args) throws Exception {/*** 获取ExecutionEnvironment运行环境* 使用readTextFile读取数据构建数据源* 创建一个ApacheLogEvent类* 使用map操作执行转换* 打印测试*///TODO 获取ExecutionEnvironment运行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//TODO 使用readTextFile读取数据构建数据源DataStream<String> lines = env.readTextFile("./data/input/apache2.log");//TODO 创建一个ApacheLogEvent类//TODO 使用map操作执行转换/*** String：传入值类型* ApacheEvent：返回值类型*/SingleOutputStreamOperator<ApacheEvent> apacheEventBean = lines.map(new MapFunction<String, ApacheEvent>() {@Overridepublic ApacheEvent map(String line) throws Exception {String[] elements = line.split(" ");String ip = elements[0];int userId = Integer.parseInt(elements[1]);SimpleDateFormat simpleDateFormat = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss");long timestamp = simpleDateFormat.parse(elements[2]).getTime();String method = elements[3];String path = elements[4];return new ApacheEvent(ip, userId, timestamp, method, path);}});//TODO 打印测试apacheEventBean.print();env.execute();}@Data@AllArgsConstructor@NoArgsConstructorpublic static class ApacheEvent{String ip;      // 访问ipint userId;     // 用户idlong timestamp; // 访问时间戳String method;  // 访问方法String path;    // 访问路径}
}

2: flatMap

将一各元素转换成多个元素，压扁操作

package cn.itcast.flink.transformation;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;public class FlatMapDemo {public static void main(String[] args) throws Exception {/*** 开发步骤：* 构建批处理运行环境* 构建本地集合数据源* 使用flatMap将一条数据经过处理转换为三条数据* 使用逗号分隔字段* 分别构建三条数据* 打印输出*///TODO 构建批处理运行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//TODO 构建本地集合数据源DataStream<String> lines = env.readTextFile("./data/input/flatmap.log");//TODO 使用flatMap将一条数据经过处理转换为三条数据//TODO 使用逗号分隔字段//TODO 分别构建三条数据SingleOutputStreamOperator<String> result = lines.flatMap(new FlatMapFunction<String, String>() {@Overridepublic void flatMap(String line, Collector<String> collector) throws Exception {String[] elements = line.split(",");collector.collect(elements[0] + "有" + elements[1]);collector.collect(elements[0] + "有" + elements[2]);collector.collect(elements[0] + "有" + elements[3]);}});//TODO 打印输出result.print();env.execute();}
}

3:filter

将满足条件的数据过滤出来

Demo1:

package cn.itcast.flink.transformation;import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class FilterDemo {public static void main(String[] args) throws Exception {/*** 获取ExecutionEnvironment运行环境* 使用fromCollection构建数据源* 使用filter操作执行过滤* 打印测试*///TODO 获取ExecutionEnvironment运行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//TODO 使用fromCollection构建数据源DataStream<String> lines = env.readTextFile("./data/input/apache.log");//TODO 使用filter操作执行过滤(66.249.73.135)SingleOutputStreamOperator<String> result = lines.filter(new FilterFunction<String>() {@Overridepublic boolean filter(String line) throws Exception {return line.contains("66.249.73.135");}});//TODO 打印测试result.print();env.execute();}
}

 Demo2:

package cn.itcast.flink.transformation;import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** Author itcast* Date 2022/1/11 17:09* Desc TODO*/
public class Filter2Demo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<String> source = env.fromElements("abc123", "bcd", "192.168.0.1", "hello flink");SingleOutputStreamOperator<String> result = source.filter(new FilterFunction<String>() {@Overridepublic boolean filter(String value) throws Exception {return value.contains("flink");}});result.printToErr();source.filter(t->t.contains("flink")).printToErr();env.execute();}
}

 3:keyby

对流上的数据进行分流（分组）

Demo1:

package cn.itcast.flink.transformation;import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.ArrayList;
import java.util.List;/*** Author itcast* Date 2022/1/11 17:17* Desc TODO*/
public class KeyedDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//keyBy支持几种类型？ 两种 ① 元祖 Tuple ② Pojo 普通的java类/*DataStreamSource<Tuple2<String, Integer>> source = env.fromElements(Tuple2.of("男", 1), Tuple2.of("男", 1), Tuple2.of("女", 1), Tuple2.of("男", 1));*/List<User> users = new ArrayList<>();//users.add(new User("男", 1));users.add(new User("男", 1));users.add(new User("女", 1));users.add(new User("男", 1));users.add(new User("女", 1));DataStreamSource<User> source1 = env.fromCollection(users);//keyBy分组/*source.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {@Overridepublic String getKey(Tuple2<String, Integer> value) throws Exception {return value.f0;}}).sum(1).printToErr();*/source1.keyBy(new KeySelector<User, String>() {@Overridepublic String getKey(User value) throws Exception {return value.getGender();}}).reduce(new ReduceFunction<User>() {@Overridepublic User reduce(User value1, User value2) throws Exception {return new User(value2.gender, value1.num + value2.num);}}).printToErr();env.execute();}public static class User {private String gender;private Integer num;public User(String gender, int num) {this.gender = gender;this.num = num;}public String getGender() {return gender;}public void setGender(String gender) {this.gender = gender;}public int getNum() {return num;}public void setNum(int num) {this.num = num;}@Overridepublic String toString() {return "User{" +"gender='" + gender + '\'' +", num=" + num +'}';}}
}

Demo2:

package cn.itcast.flink.api.transformation;import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
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 org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;/*** Author itcast* Date 2021/11/29 16:58* Desc 基于元祖中 f0 进行分组， f1 进行求和*/
public class KeybyDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//本地获取一个数据源DataStreamSource<Tuple2<String, Integer>> source = env.fromElements(Tuple2.of("篮球", 1),Tuple2.of("篮球", 2),Tuple2.of("篮球", 3),Tuple2.of("足球", 3),Tuple2.of("足球", 2),Tuple2.of("足球", 3));//对当前的数据流进行分组操作KeyedStream<Tuple2<String, Integer>, String> keyedStream = source.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {@Overridepublic String getKey(Tuple2<String, Integer> value) throws Exception {return value.f0;}});//通过 lambda 表达式KeyedStream<Tuple2<String, Integer>, String> keyedStream1 = source.keyBy(t -> t.f0);//聚合操作SingleOutputStreamOperator<Tuple2<String, Integer>> result = keyedStream.sum(1);//打印输出result.print();//执行程序env.execute();}
}

4:reduce

reduce进行自定义的聚合操作

 

package cn.itcast.flink.api.transformation;import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** Author itcast* Date 2021/11/29 17:07* Desc Reduce 聚合操作* 将每行的数据进行切分，统计每个ip对应的page view 页面展现次数* 86.149.9.216 10001 17/05/2015:10:05:30 GET /presentations/logstash-monitorama-2013/images/github-contributions.png* 83.149.9.216 10002 17/05/2015:10:06:53 GET /presentations/logstash-monitorama-2013/css/print/paper.css* 83.149.9.216 10002 17/05/2015:10:06:53 GET /presentations/logstash-monitorama-2013/css/print/paper.css* 83.149.9.216 10002 17/05/2015:10:06:53 GET /presentations/logstash-monitorama-2013/css/print/paper.css* 83.149.9.216 10002 17/05/2015:10:06:53 GET /presentations/logstash-monitorama-2013/css/print/paper.css* 83.149.9.216 10002 17/05/2015:10:06:53 GET /presentations/logstash-monitorama-2013/css/print/paper.css* 83.149.9.216 10002 17/05/2015:10:06:53 GET /presentations/logstash-monitorama-2013/css/print/paper.css* 10.0.0.1 10003 17/05/2015:10:06:53 POST /presentations/logstash-monitorama-2013/css/print/paper.css* 10.0.0.1 10003 17/05/2015:10:07:53 POST /presentations/logstash-monitorama-2013/css/print/paper.css* 10.0.0.1 10003 17/05/2015:10:08:53 POST /presentations/logstash-monitorama-2013/css/print/paper.css* 10.0.0.1 10003 17/05/2015:10:09:53 POST /presentations/logstash-monitorama-2013/css/print/paper.css* 10.0.0.1 10003 17/05/2015:10:10:53 POST /presentations/logstash-monitorama-2013/css/print/paper.css* 10.0.0.1 10003 17/05/2015:10:16:53 POST /presentations/logstash-monitorama-2013/css/print/paper.css* 10.0.0.1 10003 17/05/2015:10:26:53 POST /presentations/logstash-monitorama-2013/css/print/paper.css*/
public class ReduceDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<String> source = env.readTextFile("D:\\project\\flinkbase26\\data\\apache.log");//通过map进行映射// String : ip  Integer : 1SingleOutputStreamOperator<Tuple2<String, Integer>> result = source.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String value) throws Exception {//todoString[] arrs = value.split(" ");String ip = arrs[0];return Tuple2.of(ip,1);}}).keyBy(t -> t.f0).reduce(new ReduceFunction<Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {return Tuple2.of(value1.f0, value1.f1 + value2.f1);}});//打印输出result.print();//exeenv.execute();}
}

5:minBy和maxBy

求极值，最大最小值

minBy/maxBy，遇到较小/较大的会显示当前这一组Tuple值，min/max 只管当前一个值，不管别的值

Demo1:

package cn.itcast.flink.transformation;import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class MinMinByDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStream<Tuple3<Integer, Integer, Integer>> source = env.fromElements(Tuple3.of(1, 3, 2),Tuple3.of(1, 1, 2),Tuple3.of(1, 2, 3),Tuple3.of(1, 111, 1),Tuple3.of(1, 1, 1),Tuple3.of(1, 2, 0),Tuple3.of(1, 33, 2));//取最小值，别的值按顺序填充source.keyBy(t -> t.f0).min(2).print("min>>>");//获取最小值，取第一个最小值的整行数据source.keyBy(t -> t.f0).minBy(2).printToErr("minBy>>>");/** min 会取最小值记录下来, 其它字段按照顺序填充, 最终将最后一个元素的其它字段和最小值组合形成结果返回* minBy 会判断最小值, 将第一个出现的最小值所在的那一行的结果保留, 最终返回出去* max maxBy同理*/env.execute();}
}

Demo2:

public class MinMinByDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//调用Source创建DataStream//spark,2//spark,5//hadoop,7//hadoop,3DataStream<String> lines = env.socketTextStream("node1", 9999);SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String value) throws Exception {String[] fields = value.split(",");String word = fields[0];int count = Integer.parseInt(fields[1]);return Tuple2.of(word, count);}});KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(w -> w.f0);keyed.minBy(1).print("最小数据>>>");keyed.maxBy(1).print("最大数据>>>");env.execute();}
}

• union 和 connect 合流算子

  • 将两个流（相同类型或不同类型合并到一起），union就是相同数据类型，connect不同数据类型

  • union 是将两个相同类型的数据放到一起，不会去重，类似于 数据库SQL中的行合并 union all

  • connect 一般两个数据流，数据流和控制流(配置流)进行合并拉宽，生辰新的数据流。

• side output

  • 分流就是将一个数据流拆分成多个数据流

  • 应用场景- 对日志的分流，对于严重迟到的数据对分流保存

  • 如何分流： split 分流，select 筛选流（过期）；可以使用侧输出流 side output tag

  • 侧输出流，主要用于数据的分流，将满足条件的数据流分出来存储，侧输出流中。

  • 案例- 对流中的数据按照奇数和偶数进行分流，并获取分流后的数据。

 Demo1:

package cn.itcast.flink.transformation;import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class UnionDemo {public static void main(String[] args) throws Exception {/*** 实现步骤：* 1）初始化flink的流处理的运行环境* 2）加载/创建数据源* 3）处理数据* 4）打印输出* 5）递交执行作业*/StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> ds1 = env.fromElements("hadoop", "hive", "flume");DataStreamSource<String> ds2 = env.fromElements("hadoop","hive","spark");DataStream<String> result = ds1.union(ds2);result.printToErr();env.execute();}
}

Demo2:

/*** Author itcast* Date 2021/11/29 17:41* Desc 对流中的数据按照奇数和偶数进行分流，并获取分流后的数据*/
public class SplitStreamDemo {public static void main(String[] args) throws Exception {//1.创建流处理环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//2.设置并行度env.setParallelism(1);//3.加载本地集合DataStreamSource<Long> source = env.fromSequence(1, 20);//4.数据分流,分为奇数和偶数//定义一个保存奇数的侧输出流OutputTag<Long> odd = new OutputTag<>("odd",Types.LONG);OutputTag<Long> even = new OutputTag<>("even",Types.LONG);SingleOutputStreamOperator<Long> evenStream = source.process(new ProcessFunction<Long, Long>() {@Overridepublic void processElement(Long value, Context ctx, Collector<Long> out) throws Exception {if (value % 2 == 0) {/*out.collect(value);*/ctx.output(even,value);} else {ctx.output(odd, value);}}});//5.获取分流后的数据evenStream.print("even:");evenStream.getSideOutput(odd).print("odd:");evenStream.getSideOutput(even).print("even: sideoutput:");//6.打印数据//7.执行任务env.execute();}
}