前言
??????昨天折腾了下SpringBoot与Flink集成,实际上集成特简单,主要是部署打包的问题折腾了不少时间。想打出的包直接可以java -jar运行,同时也可以flink run运行,或者在flink的dashboard上上传点击启动。结果是不行,但是使用不同的插件打包还是可以的。
一、SpringBoot集成Flink
??????其实没什么特别的,就把Flink依赖的包在pom引入就行了。只是FlinkTask的写法要小调整下,把相关依赖交给spring管理就行。 ??????然后如果放弃Flink的Dashboard端监控task执行相关信息,那也可以在SpringBoot的启动类里调用就行,但是可能出现task的相关对象没有注入,这种都是小问题(实际就是springboot启动完成再调用,或者通过自动任务调用。也可以在springBoot的入口类用@ComponentScan注解扫描flinkTask所在的目录)。 ??????实际更潇洒一点的做法可以将flinkTask的信息存表,通过springBoot的接口调用restfullApi接口,接口里调用task,甚至可以做task的启停、线程监控(接口里开线程调用task)。
二、FlinkTask写法调整
@Component
@Slf4j
public class JianGongStopCarTask {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<JGParkingLotInfo> dataStream = env.addSource(new JGParkingLotInfoSource());
SingleOutputStreamOperator<JGParkingLotInfo> jgParkingLotInfoSingleOutputStreamOperator = dataStream.map(jgParkingLotInfo -> jgParkingLotInfo);
jgParkingLotInfoSingleOutputStreamOperator.addSink(new SinkToMySQL());
env.execute();
}
}
PS: 实际就是@Component这个注解,然后这个示例里的JGParkingLotInfoSource、SinkToMySQL类都需要加这个注解。其他就没有什么调整了,然后如果还有其他依赖没有,使用hutool的SpringUtil进行get对象就行。
三、打包插件
<plugins>
<!-- 编译插件 -->
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.8</source>
<target>1.8</target>
<encoding>UTF-8</encoding>
</configuration>
</plugin>
<!-- spring boot 项目打包
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>-->
<!-- Flink打包方式一 -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.3.0</version>
<configuration>
<archive>
<manifest>
<mainClass>com.easylinkin.dc.olap.JianGongStopCarTask</mainClass>
</manifest>
</archive>
<!-- 打包依赖 -->
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
<!-- flink打包方式二
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.3.0</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<createDependencyReducedPom>false</createDependencyReducedPom>
<artifactSet>
<excludes>
<exclude>com.google.code.findbugs:jsr305</exclude>
<exclude>org.slf4j:*</exclude>
<exclude>log4j:*</exclude>
</excludes>
</artifactSet>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>module-info.class</exclude>
<exclude>META-INF
除了打成SpringBoot用springboot的插件打包,flinkTask的打包有3种方式,方式三适合lib包提前传到task指定的依赖存储目录。这样上传flinkTask就很小。 方式二是官方推荐FlinkTask的打包方式,地址:https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/configuration/maven/ 说一千道一万就是因为打包的META-INF下的MANIFEST.MF文件的内容有区别。springBoot项目的这个文件有自己的JarLauncher。 FlinkTask的jar这个文件内容
四、Flink的上传与运行
1、上传并命令运行 配置好flink环境,命令就是
flink run WordCount.jar
2、Flink管理大屏上传运行 点击“Submit”
总结
- 这样处理应该是最优雅的了,task的写法改动也小。
- 官方用的是方式二打包,实际我觉得要是依赖特多,用方式三打包,然后将依赖的jar传到flink运行环境flinkTask指定的目录下应该也不错(flinkTask的包就变小了)
- flink的这个计算监控真香
??????所以基本还是就当springBoot集成flink的项目一样开发吧,在打包的时候注意切换插件就ok了。就分享到这里,up!
|