Flink-02 Flink Java 3分钟上手 Stream SingleOutputStreamOpe ExecutionEnvironment DataSet FlatMapFunction
CSDN 2024-10-12 09:05:03 阅读 79
代码仓库
会同步代码到 GitHub
https://github.com/turbo-duck/flink-demo
接着上一节的内容
https://blog.csdn.net/w776341482/article/details/139873938
pom内容
<code><?xml version="1.0" encoding="UTF-8"?>code>
<project xmlns="http://maven.apache.org/POM/4.0.0"code>
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"code>
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">code>
<modelVersion>4.0.0</modelVersion>
<groupId>org.example</groupId>
<artifactId>flink-demo-01</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<flink.version>1.13.2</flink.version> <!-- 确保版本号正确 -->
<scala.binary.version>2.12</scala.binary.version> <!-- 确保Scala版本正确 -->
</properties>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>
</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
</dependencies>
</project>
编写代码
定义变量
String ip = "0.0.0.0";
int port = 9999;
获取环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Socket流
DataStreamSource<String> textStream = streamExecutionEnvironment.socketTextStream(ip, port, "\n");
FlatMap
SingleOutputStreamOperator<Tuple2<String, Long>> tuple2SingleOutputStreamOperator = textStream.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
@Override
public void flatMap(String s, Collector<Tuple2<String, Long>> collector) throws Exception {
String[] splits = s.split("\\s");
for (String word : splits) {
collector.collect(Tuple2.of(word, 1L));
}
}
});
Stream
SingleOutputStreamOperator<Tuple2<String, Long>> word = tuple2SingleOutputStreamOperator
.keyBy(new KeySelector<Tuple2<String, Long>, Object>() {
@Override
public Object getKey(Tuple2<String, Long> stringLongTuple2) throws Exception {
return stringLongTuple2.f0;
}
})
.window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(1)))
.sum(1);
完整代码
package icu.wzk.demo02;
import org.apache.flink.api.common.functions.FlatMapFunction;
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.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
public class StartApp {
public static void main(String[] args) throws Exception {
String ip = "0.0.0.0";
int port = 9999;
StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> textStream = streamExecutionEnvironment.socketTextStream(ip, port, "\n");
SingleOutputStreamOperator<Tuple2<String, Long>> tuple2SingleOutputStreamOperator = textStream.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
@Override
public void flatMap(String s, Collector<Tuple2<String, Long>> collector) throws Exception {
String[] splits = s.split("\\s");
for (String word : splits) {
collector.collect(Tuple2.of(word, 1L));
}
}
});
SingleOutputStreamOperator<Tuple2<String, Long>> word = tuple2SingleOutputStreamOperator
.keyBy(new KeySelector<Tuple2<String, Long>, Object>() {
@Override
public Object getKey(Tuple2<String, Long> stringLongTuple2) throws Exception {
return stringLongTuple2.f0;
}
})
.window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(1)))
.sum(1);
word.print();
streamExecutionEnvironment.execute("stream!");
}
}
启动服务
启动一个服务，等会儿用作给 Flink 服务发送数据。
Mac平台
nc -lk 9999
Win平台
telnet 127.0.0.1 9999
运行效果
启动Flink服务，在刚才启动的Shell控制台中，快速的输入一些数字并回车。观察效果
此时可以看到Flink的控制台有了对应的响应：
<code>SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
4> (1,1)
4> (1,4)
2> (2,2)
4> (1,5)
4> (1,5)
2> (2,4)
2> (2,4)
3> (3,2)
4> (1,5)
1> (4,1)
2> (2,4)
3> (3,3)
4> (1,4)
1> (4,3)
3> (3,4)
4> (1,3)
1> (4,3)
2> (2,4)
2> (2,2)
3> (3,4)
4> (1,2)
1> (4,3)
3> (3,4)
4> (1,2)
1> (4,3)
3> (3,2)
4> (1,2)
1> (4,2)
3> (3,1)
4> (1,2)
上一篇： 双指针算法的一个简单题解
下一篇： [Qt] 信号与槽：深入浅出跨UI与跨线程的信号发送
本文标签
Flink-02 Flink Java 3分钟上手 Stream SingleOutputStreamOpe ExecutionEnvironment DataSet FlatMapFunction
声明
本文内容仅代表作者观点，或转载于其他网站，本站不以此文作为商业用途
如有涉及侵权，请联系本站进行删除
转载本站原创文章，请注明来源及作者。