SpringBoot集成flink

Flink是一个批处理和流处理结合的统一计算框架，其核心是一个提供了数据分发以及并行化计算的流数据处理引擎。
最大亮点是流处理，最适合的应用场景是低时延的数据处理。
场景：高并发pipeline处理数据，时延毫秒级，且兼具可靠性。
环境搭建：
①、安装flink
https://nightlies.apache.org/flink/flink-docs-master/zh/docs/try-flink/local_installation/
②、安装Netcat
Netcat（又称为NC）是一个计算机网络工具，它可以在两台计算机之间创建 TCP/IP 或 UDP 连接。
用于测试网络中的端口，发送文件等操纵。
举行网络调试和探测，也可以举行加密连接和长途管理等高级网络操纵

1. yum install -y nc # 安装nc命令
3. nc -lk 8888 # 启动socket端口

复制代码

无界流之读取socket文本流

一、依靠

2. <?xml version="1.0" encoding="UTF-8"?>
3. <project xmlns="http://maven.apache.org/POM/4.0.0"
4.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
5.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
6.     <parent>
7.         <artifactId>springboot-demo</artifactId>
8.         <groupId>com.et</groupId>
9.         <version>1.0-SNAPSHOT</version>
10.     </parent>
11.     <modelVersion>4.0.0</modelVersion>
13.     <artifactId>flink</artifactId>
15.     <properties>
16.         <maven.compiler.source>8</maven.compiler.source>
17.         <maven.compiler.target>8</maven.compiler.target>
18.     </properties>
19.     <dependencies>
21.         <dependency>
22.             <groupId>org.springframework.boot</groupId>
23.             <artifactId>spring-boot-starter-web</artifactId>
24.         </dependency>
26.         <dependency>
27.             <groupId>org.springframework.boot</groupId>
28.             <artifactId>spring-boot-autoconfigure</artifactId>
29.         </dependency>
30.         <dependency>
31.             <groupId>org.springframework.boot</groupId>
32.             <artifactId>spring-boot-starter-test</artifactId>
33.             <scope>test</scope>
34.         </dependency>
35.         <!-- 添加 Flink 依赖 -->
36.         <dependency>
37.             <groupId>org.apache.flink</groupId>
38.             <artifactId>flink-streaming-java</artifactId>
39.             <version>1.17.0</version>
40.         </dependency>
41.         <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-java -->
42.         <dependency>
43.             <groupId>org.apache.flink</groupId>
44.             <artifactId>flink-java</artifactId>
45.             <version>1.17.0</version>
46.         </dependency>
48.         <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients -->
49.         <dependency>
50.             <groupId>org.apache.flink</groupId>
51.             <artifactId>flink-clients</artifactId>
52.             <version>1.17.0</version>
53.         </dependency>
55.         <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-base -->
56.         <dependency>
57.             <groupId>org.apache.flink</groupId>
58.             <artifactId>flink-connector-base</artifactId>
59.             <version>1.17.0</version>
60.         </dependency>
62.         <dependency>
63.             <groupId>org.apache.flink</groupId>
64.             <artifactId>flink-connector-files</artifactId>
65.             <version>1.17.0</version>
66.         </dependency>
67.         <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka -->
68.         <dependency>
69.             <groupId>org.apache.flink</groupId>
70.             <artifactId>flink-connector-kafka</artifactId>
71.             <version>1.17.0</version>
72.         </dependency>
73.         <dependency>
74.             <groupId>org.apache.flink</groupId>
75.             <artifactId>flink-runtime-web</artifactId>
76.             <version>1.17.0</version>
77.         </dependency>
80.     </dependencies>
81.     <build>
82.         <plugins>
83.             <plugin>
84.                 <groupId>org.apache.maven.plugins</groupId>
85.                 <artifactId>maven-shade-plugin</artifactId>
86.                 <executions>
87.                     <execution>
88.                         <phase>package</phase>
89.                         <goals>
90.                             <goal>shade</goal>
91.                         </goals>
92.                         <configuration>
93.                             <transformers>
94.                                 <transformer
95.                                         implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
96.                                     <resource>META-INF/spring.handlers</resource>
97.                                 </transformer>
98.                                 <transformer
99.                                         implementation="org.springframework.boot.maven.PropertiesMergingResourceTransformer">
100.                                     <resource>META-INF/spring.factories</resource>
101.                                 </transformer>
102.                                 <transformer
103.                                         implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
104.                                     <resource>META-INF/spring.schemas</resource>
105.                                 </transformer>
106.                                 <transformer
107.                                         implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
108.                                 <transformer
109.                                         implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
110.                                     <mainClass>com.et.flink.job.SocketJob</mainClass>
111.                                 </transformer>
112.                             </transformers>
113.                         </configuration>
114.                     </execution>
115.                 </executions>
116.             </plugin>
117.         </plugins>
118.     </build>
120. </project>

复制代码

二、SoketJob

1. public class SocketJob{
2.        
3.         public static void main(String[] args)throws Exception{
4.                
5.                 // 创建执行环境
6.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
7.         // 指定并行度,默认电脑线程数
8.         env.setParallelism(3);
9.         // 读取数据socket文本流 指定监听 IP 端口 只有在接收到数据才会执行任务
10.         DataStreamSource<String> socketDS = env.socketTextStream("172.24.4.193", 8888);
12.         // 处理数据: 切换、转换、分组、聚合 得到统计结果
13.         SingleOutputStreamOperator<Tuple2<String, Integer>> sum = socketDS
14.                 .flatMap(
15.                         (String value, Collector<Tuple2<String, Integer>> out) -> {
16.                             String[] words = value.split(" ");
17.                             for (String word : words) {
18.                                 out.collect(Tuple2.of(word, 1));
19.                             }
20.                         }
21.                 )
22.                 .setParallelism(2)
23.                 // // 显式地提供类型信息:对于flatMap传入Lambda表达式，系统只能推断出返回的是Tuple2类型，而无法得到Tuple2<String, Long>。只有显式设置系统当前返回类型，才能正确解析出完整数据
24.                 .returns(new TypeHint<Tuple2<String, Integer>>() {
25.                 })
26. //                .returns(Types.TUPLE(Types.STRING,Types.INT))
27.                 .keyBy(value -> value.f0)
28.                 .sum(1);
31.         // 输出
32.         sum.print();
34.         // 执行
35.         env.execute();
36.         }
37. }

复制代码

测试：
启动socket流：

1. nc -l 8888

复制代码

本地执行：直接ideal启动main步伐，在socket流中输入

1. abc bcd cde
2. bcd cde fgh
3. cde fgh hij

复制代码

集群执行：
执行maven打包，将打包的jar上传到集群中

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。