SpringBoot集成flink全过程
更新时间：2025年01月14日 14:45:36 作者：A尘埃
文章介绍了Flink作为批处理和流处理结合的统一计算框架,特别是其强大的流处理能力,文章还详细描述了如何在本地和集群环境中搭建Flink,并通过Netcat工具生成一个无界流测试,文章最后提供了启动socket流的步骤和示例代码,希望对读者有所帮助
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 连接。
用于测试网络中的端口，发送文件等操作。
进行网络调试和探测，也可以进行加密连接和远程管理等高级网络操作
yum install -y nc # 安装nc命令 nc -lk 8888 # 启动socket端口
无界流之读取socket文本流
一、依赖
<?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">
<parent>
<artifactId>springboot-demo</artifactId>
<groupId>com.et</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>flink</artifactId>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-autoconfigure</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!-- 添加 Flink 依赖 -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
<version>1.17.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-java -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.17.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
<version>1.17.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-base -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-base</artifactId>
<version>1.17.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-files</artifactId>
<version>1.17.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka</artifactId>
<version>1.17.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime-web</artifactId>
<version>1.17.0</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<transformers>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/spring.handlers</resource>
</transformer>
<transformer
implementation="org.springframework.boot.maven.PropertiesMergingResourceTransformer">
<resource>META-INF/spring.factories</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/spring.schemas</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
<transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>com.et.flink.job.SocketJob</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>二、SoketJob
public class SocketJob{
public static void main(String[] args)throws Exception{
// 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 指定并行度,默认电脑线程数
env.setParallelism(3);
// 读取数据socket文本流 指定监听 IP 端口 只有在接收到数据才会执行任务
DataStreamSource<String> socketDS = env.socketTextStream("172.24.4.193", 8888);
// 处理数据: 切换、转换、分组、聚合 得到统计结果
SingleOutputStreamOperator<Tuple2<String, Integer>> sum = socketDS
.flatMap(
(String value, Collector<Tuple2<String, Integer>> out) -> {
String[] words = value.split(" ");
for (String word : words) {
out.collect(Tuple2.of(word, 1));
}
}
)
.setParallelism(2)
// // 显式地提供类型信息:对于flatMap传入Lambda表达式，系统只能推断出返回的是Tuple2类型，而无法得到Tuple2<String, Long>。只有显式设置系统当前返回类型，才能正确解析出完整数据
.returns(new TypeHint<Tuple2<String, Integer>>() {
})
// .returns(Types.TUPLE(Types.STRING,Types.INT))
.keyBy(value -> value.f0)
.sum(1);
// 输出
sum.print();
// 执行
env.execute();
}
}测试：
启动socket流：
nc -l 8888
本地执行：直接ideal启动main程序，在socket流中输入
abc bcd cde bcd cde fgh cde fgh hij
集群执行：
执行maven打包，将打包的jar上传到集群中
总结
以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。
相关文章
Spring Initializr只能创建为Java 17版本以上的问题解决
这篇文章主要给大家介绍了关于Spring Initializr只能创建为Java 17版本以上问题的解决办法,文中通过图文介绍的非常详细,对大家的学习或者工作具有一定的参考借鉴价值,需要的朋友可以参考下2024-01-01
Spring三级缓存机制解决了循环依赖问题,通过一级缓存存放完全初始化的bean,二级缓存存放实例化但未完成依赖注入和初始化的bean,三级缓存存放bean的创建工厂,避免了重复创建和确保代理对象的正确生成2025-02-02
这篇文章主要介绍了Tomcat调优方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2024-04-04
这篇文章主要介绍了JAVA异常和自定义异常处理方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2023-09-09
Apache Shiro是一个强大且灵活的开源安全框架,适用于Java应用程序,它提供了认证、授权、会话管理和加密等功能,通过Shiro,开发者可以轻松地实现用户身份验证和权限控制,从而保护应用程序的安全,本文介绍Springboot中的Shiro基础入门教程,感兴趣的朋友跟随小编一起看看吧2025-12-12
这篇文章主要为大家详细介绍了Servlet实现简单文件上传功能，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2017-10-10
浅谈web服务器项目中request请求和response的相关响应处理
这篇文章主要介绍了浅谈web服务器项目中request请求和response的相关响应处理，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2020-07-07
本文介绍了在Spring Boot中创建和使用自定义注解的方法,通过自定义注解,可以减少重复代码、增强代码可读性和可维护性,具体步骤包括定义注解、创建注解处理器以及在业务方法上使用注解2025-11-11
这篇文章主要介绍了别在Java代码里乱打日志了，这才是正确的打日志姿势，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2019-06-06
Java迭代器是遍历Collection集合的标准工具,提供hasNext()、next()和remove()三个核心方法,下面就来介绍一下Collection迭代器的实现,感兴趣的可以了解一下2026-01-01
最新评论