时间:2023-01-17 12:47:13 | 栏目:JAVA代码 | 点击:次
其实没什么特别的,就把Flink依赖的包在pom引入就行了。只是FlinkTask的写法要小调整下,把相关依赖交给spring管理就行。
然后如果放弃Flink的Dashboard端监控task执行相关信息,那也可以在SpringBoot的启动类里调用就行,但是可能出现task的相关对象没有注入,这种都是小问题(实际就是springboot启动完成再调用,或者通过自动任务调用。也可以在springBoot的入口类用@ComponentScan注解扫描flinkTask所在的目录)。
实际更潇洒一点的做法可以将flinkTask的信息存表,通过springBoot的接口调用restfullApi接口,接口里调用task,甚至可以做task的启停、线程监控(接口里开线程调用task)。
@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/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> </excludes> </filter> </filters> <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer"> <resource>META-INF/spring.handlers</resource> <resource>reference.conf</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.easylinkin.dc.olap.JianGongStopCarTask</mainClass> </transformer> </transformers> </configuration> </execution> </executions> </plugin> --> <!-- flink打包方式三 <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-jar-plugin</artifactId> <configuration> <archive> <manifest> <mainClass>com.easylinkin.dc.olap.JianGongStopCarTask</mainClass> <useUniqueVersions>false</useUniqueVersions> <addClasspath>true</addClasspath> <classpathPrefix>./lib/</classpathPrefix> </manifest> </archive> <excludes> <exclude>module-info.class</exclude> <exclude>META-INF/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> <exclude>com.google.code.findbugs:jsr305</exclude> <exclude>org.slf4j:*</exclude> <exclude>log4j:*</exclude> </excludes> </configuration> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-dependency-plugin</artifactId> <version>3.3.0</version> <executions> <execution> <id>copy-dependencies</id> <phase>package</phase> <goals> <goal>copy-dependencies</goal> </goals> <configuration> <outputDirectory>${project.build.directory}/lib</outputDirectory> <excludeTransitive>false</excludeTransitive> <stripVersion>false</stripVersion> </configuration> </execution> </executions> </plugin>--> </plugins>
除了打成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这个文件内容
1、上传并命令运行
配置好flink环境,命令就是
flink run WordCount.jar
2、Flink管理大屏上传运行
点击“Submit”
这样处理应该是最优雅的了,task的写法改动也小。
官方用的是方式二打包,实际我觉得要是依赖特多,用方式三打包,然后将依赖的jar传到flink运行环境flinkTask指定的目录下应该也不错(flinkTask的包就变小了)
flink的这个计算监控真香
所以基本还是就当springBoot集成flink的项目一样开发吧,在打包的时候注意切换插件就ok了。就分享到这里,up!