代码仓库
会同步代码到 GitHub
https://github.com/turbo-duck/flink-demo
接着上一节的内容
https://blog.csdn.net/w776341482/article/details/139873938
pom内容
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <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的控制台有了对应的响应：
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)
