版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
大数据系列文章目录
官方网址：https://flink.apache.org/
学习资料：https://flink-learning.org.cn/
目录
- 官网所有的Transformation操作
- KeyBy
- Reduce
- Aggregations
- Union
- Connect
- Split和select
官网所有的Transformation操作
DataStream包括一系列的Transformation操作：
https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/stream/operators/index.html
KeyBy
按照指定的key来进行分流，类似于SQL中的groupBy。
示例
自定义数据源, 进行单词的计数
开发步骤
- 获取流处理运行环境
- 设置并行度
- 获取数据源
- 转换操作
(1) 以空白进行分割
(2) 给每个单词计数1
(3) 根据单词分组 - 求和
- 打印到控制台
- 执行任务
package batch.transformation;
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 lwh
* @date 2023/4/25
* @description 演示keyBy方法 也使用了aggregate方法的sum求和
**/
public class KeyByDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 设置全局并行度为1
env.setParallelism(1);
// Get source
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)
);
// 分组加聚合计算, 类似SQL的group by 后加聚合函数求每个组的数据
// 有一点要注意的是, SQL中是对分组后的每个组的全量数据做聚合计算, 是批计算
// 在流计算内,是来一条计算一条,也就是每个组的数据,挨个进行计算,求和累加,所以结果中最后一个打印的数据才是最终的求和结果
SingleOutputStreamOperator<Tuple2<String, Integer>> sum = source.keyBy(0).sum(1);
// 如果不分组的话, sum的结果是 1+2+3+3+2+3 = 14 分组后是 篮球 6 足球 8
sum.print();
env.execute();
}
}
Reduce
可以对一个datastream或者一个group来进行聚合计算，最终聚合成一个元素
数据源
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
10.0.0.1 10003 17/05/2015:10:26:53 POST /presentations/logstash-monitorama-2013/css/print/paper.css
示例
读取本地文件，根据IP地址统计计数
步骤
- 获取ExecutionEnvironment运行环境
- 使用fromCollection构建数据源
- 使用reduce执行聚合操作
- 打印测试
package batch.transformation;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.operators.ReduceOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
/**
* @author lwh
* @date 2023/4/25
* @description
**/
public class ReduceDemo {
public static void main(String[] args) throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
DataSource<String> logSource = env.readTextFile("data/input/apache.log");
// 提取IP, 后面都跟上1(作为元组返回)
MapOperator<String, Tuple2<String, Integer>> ipWithOne = logSource.map(new MapFunction<String, Tuple2<String, Integer>>() {
@Override
public Tuple2<String, Integer> map(String value) throws Exception {
String ip = value.split(" ")[0];
return Tuple2.of(ip, 1);
}
});
// 分组 + reduce聚合
UnsortedGrouping<Tuple2<String, Integer>> grouped = ipWithOne.groupBy(0);
// reduce 聚合
ReduceOperator<Tuple2<String, Integer>> result = grouped.reduce(new ReduceFunction<Tuple2<String, Integer>>() {
@Override
public 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();
}
}
Aggregations
按照内置的方式来进行聚合。例如：SUM/MIN/MAX…
示例
请将以下元组数据，使用aggregate操作进行单词统计
("java" , 1) , ("java", 1) ,("scala" , 1)
步骤
- 获取StreamExecutionEnvironment运行环境
- 使用fromCollection构建数据源
- 使用groupBy按照单词进行分组
- 使用aggregate对每个分组进行SUM统计
- 打印测试
package batch.transformation;
import org.apache.flink.api.java.tuple.Tuple;
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 java.util.Arrays;
/**
* @author lwh
* @date 2023/4/25
* @description 使用内置的方式进行聚合数据 , aggregate只能将数据作用到元组上，例如：sum\Max\min
**/
public class StreamAggregateDemo {
public static void main(String[] args) throws Exception {
//1：获取ExecutionEnvironment运行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//2： 使用`fromCollection`构建数据源
DataStreamSource<Tuple2<String, Integer>> tupleDataStream = env.fromCollection(
Arrays.asList(Tuple2.of("java", 1), Tuple2.of("java", 1), Tuple2.of("scala", 1)));
//3. 使用groupBy按照单词进行分组
KeyedStream<Tuple2<String, Integer>, Tuple> keyByStream = tupleDataStream.keyBy(0);
//4. 使用aggregate对每个分组进行SUM统计
SingleOutputStreamOperator<Tuple2<String, Integer>> resultDataStream = keyByStream.sum( 1);
resultDataStream.printToErr();
env.execute();
}
}
Union
将两个DataStream取并集，不会去重。
两个DS的泛型需要一致
示例
将以下数据进行取并集操作
数据集1
"hadoop", "hive", "flume"
数据集2
"hadoop", "hive", "spark"
步骤
- 构建流处理运行环境
- 使用fromCollection创建两个数据源
- 使用union将两个数据源关联在一起
- 打印测试
package batch.transformation;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
/**
* @author lwh
* @date 2023/4/25
* @description 合并多个流，类型必须要一致
**/
public class UnionDemo {
public static void main(String[] args) throws Exception {
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
DataSource<String> source1 = env.fromElements("hadoop", "hive", "flume");
DataSource<String> source2 = env.fromElements("yarn", "hive", "spark");
source1.union(source2).print();
/*
Union算子 会进行合并, 不会进行重复判断
Union算子 必须进行 同类型元素的合并, 哪怕是顶级类Object也不行, 必须是实体类(撇除继承关系)的类型一致才可以
*/
}
}
Connect
DataStream,DataStream → ConnectedStreams：连接两个保持他们类型的数据流，两个数据流被 Connect 之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立, 作为对比，Union后是真的变成一个流了。
示例
读取两个不同类型的数据源，使用connect进行合并打印。
开发步骤
- 创建流式处理环境
- 添加两个自定义数据源
- 使用connect合并两个数据流,创建ConnectedStreams对象
- 遍历ConnectedStreams对象,转换为DataStream
- 打印输出,设置并行度为1
- 执行任务
package batch.transformation;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import java.util.Random;
/**
* @author lwh
* @date 2023/4/25
* @description connect连接2个流，流类型可以不一致
**/
public class ConnectDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Integer> source1 = env.addSource(new IncreaseByOneSource());
DataStreamSource<Integer> source2 = env.addSource(new IncreaseByOneSource());
ConnectedStreams<Integer, Integer> connectedStreams = source1.connect(source2);
/*
针对connect后的Stream, 所有方法都会传入特殊的处理, 比如：
map方法需要实现CoMapFunction 而非我们前面用的MapFunction
这一类特殊的处理, 比如CoMapFunction, 内就有两个map方法
1个处理连接的第一个Stream
另一个处理连接的另一个Stream
*/
// CoMapFunction的3个泛型: 泛型1: 第一个Stream的输入类型 泛型2: 第二个Stream的输入类型 泛型3: 返回的类型(要求两个map方法的返回值类型一致)
SingleOutputStreamOperator<String> map = connectedStreams.map(new CoMapFunction<Integer, Integer, String>() {
@Override
public String map1(Integer value) throws Exception {
return "map1:" + value;
}
// map1和map2的返回值类型需要一致, 因为类的第三个泛型约束着
@Override
public String map2(Integer value) throws Exception {
return "map2:" + value;
}
});
map.print("》》》》》》");
System.out.println();
env.execute();
}
static class IncreaseByOneSource implements SourceFunction<Integer> {
private boolean isRun = true; // Flag for run
private final Random random = new Random();
@Override
public void run(SourceContext<Integer> ctx) throws Exception {
while (isRun) {
ctx.collect(random.nextInt(999));
Thread.sleep(1000L);
}
}
@Override
public void cancel() {
this.isRun = false;
}
}
}
Split和select
Split就是将一个DataStream分成两个或者多个DataStream
Select就是获取分流后对应的数据
简单认为就是, Split会给数据打上标记
然后通过Select, 选择标记来划分出不同的Stream
效果类似KeyBy分流，但是比KeyBy更自由些，可以自由打标记并进行分流。
示例
加载本地集合(1,2,3,4,5,6), 使用split进行数据分流,分为奇数和偶数. 并打印奇数结果
开发步骤
- 创建流处理环境
- 设置并行度
- 加载本地集合
- 数据分流,分为奇数和偶数
- 获取分流后的数据
- 打印数据
- 执行任务
package batch.transformation;
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;
import java.util.ArrayList;
/**
* @author lwh
* @date 2023/4/25
* @description 使用split分流打标记，select在根据标记获取分流后的数据
**/
public class SplitSelectDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<Integer> source = env.fromElements(1, 2, 3, 4, 5, 6);
// 通过Split + Select的组合, 区分出偶数和奇数
// SplitStream被标记为弃用了, 但是源码中没有相关解释, 可能以后的新版本会有对应的替代对象, 目前先继续使用
SplitStream<Integer> splitedStream = source.split(new OutputSelector<Integer>() {
// 可以看出返回值是Iterable<String> 也就是泛型是String的可迭代集合
// 意思就是将字符串标记放入可迭代集合中返回即可
@Override
public Iterable<String> select(Integer value) {
ArrayList<String> flagList = new ArrayList<>();
if (value % 2 == 0) {
flagList.add("even"); // 偶数标记为even
} else {
flagList.add("odd"); // 奇数标记为odd
flagList.add("jishu"); // 标记可以打多个, 比如奇数就打了两个标记
}
return flagList; // 可以看出, 返回的是flag, 和value没关系
// 这个方法就是纯用来打标记的, 对value不做处理
}
});
// 偶数DataStream
DataStream<Integer> evenStream = splitedStream.select("even");
// 通过Select取得奇数DataStream
DataStream<Integer> oddStream = splitedStream.select("odd");
// 通过奇数的另一个标记取得奇数内容
DataStream<Integer> odd2Stream = splitedStream.select("jishu");
// 如果select的内容不存在
DataStream<Integer> hahaha = splitedStream.select("hahaha");
evenStream.print("even>>>");
oddStream.print("odd>>>");
odd2Stream.print("odd2>>>");
// hahaha标记不存在, 所以无法得到内容, 这个DataStream是空的
hahaha.print("hahaha>>>");
env.execute();
}
}
