小编给大家分享一下Flink中Transform怎么用，相信大部分人都还不怎么了解，因此分享这篇文章给大家参考一下，希望大家阅读完这篇文章后大有收获，下面让我们一起去了解一下吧！
计算
安全
数据库
网络和加速
企业服务
String path = "E:\\GIT\\flink-learn\\flink-learn\\telemetering.txt";
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
TupleTypeInfo<Tuple3<String, Double, Long>> typeInfo = new TupleTypeInfo<>(Types.STRING, Types.DOUBLE, Types.LONG);
TupleCsvInputFormat<Tuple3<String, Double, Long>> tupleCsvInputFormat =
new TupleCsvInputFormat<>(new Path(path), typeInfo);
DataStreamSource<Tuple3<String, Double, Long>> dataStreamSource = env.createInput(tupleCsvInputFormat, typeInfo);
//或 DataStreamSource<Tuple2<String, Double>> dataStreamSource = env.readFile(tupleCsvInputFormat, path);
SingleOutputStreamOperator<Tuple3<String, Double, Long>> operator = dataStreamSource
.filter(Objects::nonNull)
// .map()
// .flatMap()
// .keyBy(0)
.keyBy(tuple -> tuple.f0)
.minBy(1);
// .min()
// .max(1);
// .maxBy(1, false);
// .sum(1);
// .reduce();
// .process();
operator.print().setParallelism(1);
env.execute();String path = "E:\\GIT\\flink-learn\\flink-learn\\telemetering.txt";
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
PojoTypeInfo<TelemeterDTO> typeInfo = (PojoTypeInfo<TelemeterDTO>) Types.POJO(TelemeterDTO.class);
PojoCsvInputFormat<TelemeterDTO> inputFormat = new PojoCsvInputFormat<>(new Path(path), typeInfo, new String[]{"code", "value", "timestamp"});
DataStreamSource<TelemeterDTO> dataStreamSource = env.createInput(inputFormat, typeInfo);
//分流
SplitStream<TelemeterDTO> splitStream = dataStreamSource
.split(item -> {
if (item.getValue() > 100) {
return Collections.singletonList("high");
}
return Collections.singletonList("low");
});
DataStream<TelemeterDTO> highStream = splitStream.select("high");
DataStream<TelemeterDTO> lowStream = splitStream.select("low");
//合流
ConnectedStreams<TelemeterDTO, TelemeterDTO> connectedStreams = lowStream.connect(highStream);
// DataStream<TelemeterDTO> unionDataStream = lowStream.union(highStream); //需要类型一致
SingleOutputStreamOperator<Tuple3<String, Double, Long>> operator = connectedStreams
.map(new CoMapFunction<TelemeterDTO, TelemeterDTO, Tuple3<String, Double, Long>>() {
@Override
public Tuple3<String, Double, Long> map1(TelemeterDTO value) {
return Tuple3.of(value.getCode(), value.getValue(), value.getTimestamp());
}
@Override
public Tuple3<String, Double, Long> map2(TelemeterDTO value) {
return Tuple3.of(value.getCode(), value.getValue(), value.getTimestamp());
}
});
operator.print();
env.execute();MapFunction
FilterFunction
ReduceFunction
ProcessFunction
SourceFunction
SinkFunction
富函数 包含了生命周期，及上下文相关信息，如
open() 可以在算子创建之初建立数据库连接
close() 在在算子生命结束之前关闭资源
以上是“Flink中Transform怎么用”这篇文章的所有内容，感谢各位的阅读！相信大家都有了一定的了解，希望分享的内容对大家有所帮助，如果还想学习更多知识，欢迎关注亿速云行业资讯频道！
亿速云「云服务器」，即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘，价格低至29元/月。点击查看>>
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
原文链接：https://my.oschina.net/u/2552286/blog/4680142
