devbox
知新_RL
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

【Flink】SpringBoot整合Flink并以集群方式运行，可以通过接口来动态创建执行任务，并行度可通过接口动态配置，可以和业务进行交互，灵活性极强，扩展性极高_springboot flink

作者：知新_RL | 2024-05-13 12:26:19

赞

踩

springboot flink

查阅无数资料，爬了无数个坑!!!

整体思路:把SpringBoot当成一个任务放进Flink集群中运行，并且该任务会一直运行，当其他任务需要执行时只需要调用SpringBoot的接口来动态生成任务，可以把每一个接口都当成一个任务，调用接口时Flink会根据当前环境动态创建任务并执行

注意事项: 使用 ./flink run 命令以后台运行的方式去运行打好的jar包

一、引入以下依赖

  1.      <properties>
  2.         <maven.compiler.source>8</maven.compiler.source>
  3.         <maven.compiler.target>8</maven.compiler.target>
  4.         <java.version>1.8</java.version>
  5.         <flink.version>1.13.0</flink.version>
  6.         <scala.binary.version>2.12</scala.binary.version>
  7.         <slf4j.version>1.7.30</slf4j.version>
  8.         <mysql.version>5.1.47</mysql.version>
  9.         <spring.boot.version>2.0.3.RELEASE</spring.boot.version>
  10.     </properties>
  11.  
  12.     <parent>
  13.         <groupId>org.springframework.boot</groupId>
  14.         <artifactId>spring-boot-starter-parent</artifactId>
  15.         <version>2.1.1.RELEASE</version>
  16.         <relativePath/> <!-- lookup parent from repository -->
  17.     </parent>
  18.  
  19.     <dependencies>
  20.         <dependency>
  21.             <groupId>org.springframework.boot</groupId>
  22.             <artifactId>spring-boot-starter-web</artifactId>
  23.         </dependency>
  24.         <dependency>
  25.             <groupId>org.springframework.boot</groupId>
  26.             <artifactId>spring-boot-starter-logging</artifactId>
  27.             <exclusions>
  28.                 <exclusion>
  29.                     <groupId>*</groupId>
  30.                     <artifactId>*</artifactId>
  31.                 </exclusion>
  32.             </exclusions>
  33.         </dependency>
  34.         <dependency>
  35.             <groupId>org.apache.flink</groupId>
  36.             <artifactId>flink-java</artifactId>
  37.             <version>${flink.version}</version>
  38.         </dependency>
  39.         <dependency>
  40.             <groupId>org.apache.flink</groupId>
  41.             <artifactId>flink-streaming-java_2.11</artifactId>
  42.             <version>${flink.version}</version>
  43.         </dependency>
  44.         <dependency>
  45.             <groupId>org.apache.flink</groupId>
  46.             <artifactId>flink-clients_2.11</artifactId>
  47.             <version>${flink.version}</version>
  48.         </dependency>
  49.         <dependency>
  50.             <groupId>org.apache.flink</groupId>
  51.             <artifactId>flink-runtime-web_2.11</artifactId>
  52.             <version>${flink.version}</version>
  53.         </dependency>
  54.         <dependency>
  55.             <groupId>org.springframework.boot</groupId>
  56.             <artifactId>spring-boot-starter</artifactId>
  57.             <version>${spring.boot.version}</version>
  58.         </dependency>
  59.         <dependency>
  60.             <groupId>org.apache.flink</groupId>
  61.             <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
  62.             <version>${flink.version}</version>
  63.         </dependency>
  64.         <dependency>
  65.             <groupId>org.projectlombok</groupId>
  66.             <artifactId>lombok</artifactId>
  67.             <version>1.18.8</version>
  68.             <scope>provided</scope>
  69.         </dependency>
  70.         <dependency>
  71.             <groupId>mysql</groupId>
  72.             <artifactId>mysql-connector-java</artifactId>
  73.             <version>${mysql.version}</version>
  74.         </dependency>
  75.     </dependencies>
  76.  
  77.     <build>
  78.         <finalName>flink</finalName>
  79.         <plugins>
  80.             <plugin>
  81.                 <groupId>org.apache.maven.plugins</groupId>
  82.                 <artifactId>maven-shade-plugin</artifactId>
  83.                 <version>3.2.4</version>
  84.                 <executions>
  85.                     <execution>
  86.                         <phase>package</phase>
  87.                         <goals>
  88.                             <goal>shade</goal>
  89.                         </goals>
  90.                         <configuration>
  91.                             <createDependencyReducedPom>false</createDependencyReducedPom>
  92.                             <artifactSet>
  93.                                 <excludes>
  94.                                     <exclude>com.google.code.findbugs:jsr305</exclude>
  95.                                     <exclude>org.slf4j:*</exclude>
  96.                                     <exclude>log4j:*</exclude>
  97.                                 </excludes>
  98.                             </artifactSet>
  99.                             <filters>
  100.                                 <filter>
  101.                                     <artifact>*:*</artifact>
  102.                                     <excludes>
  103.                                         <exclude>module-info.class</exclude>
  104.                                         <exclude>META-INF/*.SF</exclude>
  105.                                         <exclude>META-INF/*.DSA</exclude>
  106.                                         <exclude>META-INF/*.RSA</exclude>
  107.                                     </excludes>
  108.                                 </filter>
  109.                             </filters>
  110.                             <transformers>
  111.                                 <transformer
  112.                                         implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
  113.                                     <resource>META-INF/spring.handlers</resource>
  114.                                     <resource>reference.conf</resource>
  115.                                 </transformer>
  116.                                 <transformer
  117.                                         implementation="org.springframework.boot.maven.PropertiesMergingResourceTransformer">
  118.                                     <resource>META-INF/spring.factories</resource>
  119.                                 </transformer>
  120.                                 <transformer
  121.                                         implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
  122.                                     <resource>META-INF/spring.schemas</resource>
  123.                                 </transformer>
  124.                                 <transformer
  125.                                         implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
  126.                                 <transformer
  127.                                         implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
  128.                                     <mainClass>cn.sdata.FlinkBootApplication</mainClass>
  129.                                 </transformer>
  130.                             </transformers>
  131.                         </configuration>
  132.                     </execution>
  133.                 </executions>
  134.             </plugin>
  135.         </plugins>
  136.     </build>

二、yml配置

  1. server:
  2.   port: 10001

三、SpringBoot启动类

  1. @SpringBootApplication
  2. public class FlinkBootApplication {
  3.     public static void main(String[] args) {
  4.         SpringApplication.run(FlinkBootApplication.class,args);
  5.         //加一个死循环来保证main方法不会停止
  6.         while (true) {
  7.             Thread.sleep(30000);
  8.         }
  9.     }
  10. }

四、Controller层业务测试代码(一个流处理、一个批处理)

  1. @RequestMapping("test")
  2. @RestController
  3. @AllArgsConstructor
  4. public class TestController {
  5.  
  6.     @GetMapping("test1")
  7.     public void test(Integer parallelism) throws Exception {
  8.         //创建执行环境
  9.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  10.         
  11.         //设置并行度
  12.         env.setParallelism(parallelism);
  13.         
  14.         //读取文本流
  15.         DataStreamSource<String> source = env.socketTextStream("192.168.1.200", 8888);
  16.         SingleOutputStreamOperator<Tuple2<String, Integer>> operator = source.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
  17.             @Override
  18.             public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
  19.                 String[] splits = s.split(" ");
  20.                 for (String split : splits) {
  21.                     collector.collect(new Tuple2<>(split, 1));
  22.                 }
  23.             }
  24.         }).filter(data -> StringUtils.isNotEmpty(data.f0)).keyBy(data -> data.f0).sum(1);
  25.         //打印
  26.         operator.print();
  27.         //执行
  28.         env.execute();
  29.     }
  30.  
  31.     @GetMapping("test2")
  32.     public void test2() throws Exception {
  33.         //执行环境
  34.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  35.  
  36.         //设置并行度
  37.         env.setParallelism(1);
  38.  
  39.         //将一列元素作为数据源
  40.         DataStreamSource<Integer> integerDataStream = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12);
  41.  
  42.         //控制台打印
  43.         integerDataStream.print("int");
  44.  
  45.         //执行任务
  46.         env.execute();
  47.     }
  48. }

五、在你的服务器上开启一个nc端口

六、打包上传至你的JobManager所在的服务器

 七、进入Flink中的bin目录

八、后台方式运行jar包

 九、通过接口访问流处理方法并行度设置为4

 可以看到流处理方法所占用了4个插槽

十、通过接口访问批处理方法

可以看到该方法占用1个插槽

 稍等一会再次刷新页面

 发现该任务以执行完毕，插槽不再占用

总结:完成SpringBoot和Flink的高度整合，通过调用接口的方式来进行Flink任务的创建，SpringBoot在处理一些数据量比较大且计算量也比较大的业务场景时就可以通过Flink进行数据的处理入库，并且支持Flink集群节点可无限制扩充

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/知新_RL/article/detail/563614
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号