大道三千:最近我修Flink
目前个人理解:
处理有界,无界流的工具
FLINK:
FLINK定义:
Flink特点
Flink分层API
流的定义
有界数据流(批处理):
有界流:数据结束了,程序也就结束了
知道数据开始以及结束的地方
无界数据流:
特征:读一条,计算一条,输出一次结果
知道数据开始的地方,却不知道结束的地方
(好似长江大河,会一直一直一直产生数据)
流的状态
个人理解:(有状态流会基于内存保存之前的数据)
如果后续流的操作需要用到之前的数据,这个流时有状态的
如果后续流的操作不需要用到之前的数据,这个流是无状态的
DataSet API:有界流批处理( 已淘汰)
1:创建执行环境
2:读取流(数据)
3:将读取到的数据,转换为方便处理的格式
4:将收集到的数据进行(分组,求和,最大,最小等....)操作
//批处理方式(有界流,因为很明确的知道这个文件在哪里结束)
public class BatchWordCount {public static void main(String[] args) throws Exception {// 1. 创建执行环境ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();// 2. 从文件读取数据 按行读取(存储的元素就是每行的文本)DataSource<String> lineDS = env.readTextFile("input/words.txt");// 3. 转换数据格式FlatMapOperator<String, Tuple2<String, Long>> wordAndOne = lineDS.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {@Override //一行数据 // 数据收集器 out:相当于是一个按照 下面格式收集数据的收集器 格式=out.collect(Tuple2.of(word,1L));public void flatMap(String line, Collector<Tuple2<String, Long>> out) throws Exception {String[] words = line.split(" "); //一行数据按照" "拆分for (String word : words) { //word = 一行中的每一个字段 如果1改成2,则统计时数目会成2Tuple2<String, Long> of = Tuple2.of(word, 1L);//每个的那次都转为这种格式out.collect(of); // 收集器添加数据 (转换格式为 (循环到的字段,1L))}}});// 4. 按照 word 进行分组 按照第一个字段分组.(字段,1L),就是按照第一个字段分组(A,1),(b,1),(c,1),(d,1),(d,1) 就是按照abcd分组UnsortedGrouping<Tuple2<String, Long>> wordAndOneUG = wordAndOne.groupBy(0);// 5. 分组内聚合统计 根据第二个字段求和,即将每个分组的第二个字段相加,得到该分组的总和AggregateOperator<Tuple2<String, Long>> sum = wordAndOneUG.sum(1);// 6. 打印结果sum.print();}
}
DataStream API:流、批一体处理
转换(flatMap)、
分组(keyBy)、
求和(sum)、
执行(execute)、
读取文本(readTextFile,有界流)
1:创建流式执行环境(基于StreamExecutionEnvironment)
2:读取文件
3:转换、分组、求和,得到统计结果
4:打印输出
5:执行
//流处理方式 (有界流,因为很明确的知道这个文件在哪里结束),如果不是本地而是网络则是无界流
public class StreamWordCount {public static void main(String[] args) throws Exception {// 1. 创建流式执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 2. 读取文件DataStreamSource<String> lineStream = env.readTextFile("input/words.txt");// 3. 转换、分组、求和,得到统计结果 SingleOutputStreamOperator<Tuple2<String, Integer>> resultList = lineStream.flatM输入类型,输出类型ap(new FlatMapFunction<String, Tuple2<String, Integer>>() {@Override //当前行数据 //要返回的类型public void flatMap(String line, Collector<Tuple2<String, Integer>> list) throws Exception {String[] fields = line.split(" ");for (String field : fields) {Tuple2<String, Integer> result = Tuple2.of(field, 1);list.collect(result);}}});//分组 // 传入的数据类型() 要分组的数据类型KeyedStream<Tuple2<String, Integer>, String> gropbyDate = resultList.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {@Overridepublic String getKey(Tuple2<String, Integer> value) throws Exception {return value.f0; //这里是类型的第一位。如(hello,1),则是根据hello进行分组}});//求和。 以上一个为例子:(hello,1)分组之后,根据1索引即第二位(hello,1)的1进行求和SingleOutputStreamOperator<Tuple2<String, Integer>> sum = gropbyDate.sum(1);//打印输出sum.print();//执行env.execute();}}
// 3. 转换、分组、求和,得到统计结果SingleOutputStreamOperator<Tuple2<String, Long>> sum = lineStream.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {@Overridepublic void flatMap(String line, Collector<Tuple2<String, Long>> out) throws Exception {String[] words = line.split(" ");for (String word : words) {out.collect(Tuple2.of(word, 1L));}}}).keyBy(data -> data.f0).sum(1);
结果:
读取socket(无界流)
事件监听(环境对象.socketTextStream(IP,端口号))
备注:先启动linux 输入命令nc -lk 7777
然后启动代码监听 7777
此时linux输入的数据会被代码抓取到
备注2:跟前两个的区别就是这个是调用的socketTextStream。其他无任何区别
//监听7777端口的数据流
// 这里代码监听了 IP地址192.168.200.130 端口号7777 的操作 。ip地址那里写主机名也行
public class SocketStreamWordCount {public static void main(String[] args) throws Exception {//构建流环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//拿到数据DataStreamSource<String> lineStream = env.socketTextStream("192.168.200.130", 7777);// 转换、分组、求和,得到统计结果SingleOutputStreamOperator<Tuple2<String, Long>> convert = lineStream.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {@Overridepublic void flatMap(String line, Collector<Tuple2<String, Long>> out) throws Exception {String[] fields = line.split(" ");for (String field : fields) {Tuple2<String, Long> of = Tuple2.of(field, 1L);out.collect(of);}}});//分组KeyedStream<Tuple2<String, Long>, Object> gropBy = convert.keyBy(new KeySelector<Tuple2<String, Long>, Object>() {@Overridepublic Object getKey(Tuple2<String, Long> value) throws Exception {return value.f0;}});//求和SingleOutputStreamOperator<Tuple2<String, Long>> sum = gropBy.sum(1);//输出sum.print();//执行env.execute();}
}
SingleOutputStreamOperator<Tuple2<String, Long>> sum = lineStream.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {String[] words = line.split(" ");for (String word : words) {out.collect(Tuple2.of(word, 1L));}}).returns(Types.TUPLE(Types.STRING, Types.LONG)).keyBy(data -> data.f0).sum(1);