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

Spring Boot整合Flink_springboot整合flink

作者：我家自动化 | 2024-05-13 12:27:42

赞

踩

springboot整合flink

使用spring boot整合flink可以快速的构建起整个应用，将关注点重点放在业务逻辑的实现上。在整合的过程中遇到许多问题，最大的问题是flink流无法访问spring容器中的类，从而导致空指针异常，解决思路是在流中进行spring bean的初始化以获得ApplicationContext，进而使用其getBean方法获取类实例。

软件版本:Spring Boot 2.1.6+Flink1.6.1+JDK1.8

程序主体：

  1. @SpringBootApplication
  2. public class HadesTmsApplication implements CommandLineRunner {
  3.  
  4.  
  5.     public static void main(String[] args) {
  6.         SpringApplication application = new SpringApplication(HadesTmsApplication.class);
  7.         application.setBannerMode(Banner.Mode.OFF);
  8.         application.run(args);
  9.     }
  10.  
  11.     @Override
  12.     public void run(String... args) {
  13.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  14. FlinkKafkaConsumer010 kafkaConsumer = new FlinkKafkaConsumer010<>("topic-name"), new SimpleStringSchema(), getProperties());
  15. DataStream<String> dataStream = env.addSource(kafkaConsumer);
  16. // 此处省略处理逻辑
  17. dataStream.addSink(new MySink());
  18.  
  19.  
  20.     }
  21.  
  22. private Properties getProperties() {
  23.         Properties properties = new Properties();
  24.         properties.setProperty("bootstrap.servers", bootstrap_servers);
  25.         properties.setProperty("zookeeper.connect", zookeeper_connect);
  26.         properties.setProperty("group.id", group_id);
  27.         properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  28.         properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  29.         return properties;
  30.     }
  31. }

说明一下：因为是非web项目，所以实现CommandLineRunner接口，重写run方法。在里面编写流处理逻辑。

如果在MySink中需要使用spring容器中的类，而MySink是一个普通的类，那么是无法访问到的。会引发空指针异常。可能有人想到了ApplicationContextAware这个接口，实现这个接口获取ApplicationContext，也即是：

  1. @Component
  2. public class ApplicationContextUtil implements ApplicationContextAware, Serializable {
  3.     private static final long serialVersionUID = -6454872090519042646L;
  4.     private static ApplicationContext applicationContext = null;
  5.  
  6.     @Override
  7.     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
  8.         if (ApplicationContextUtil.applicationContext == null) {
  9.             ApplicationContextUtil.applicationContext = applicationContext;
  10.         }
  11.     }
  12.  
  13.     public static ApplicationContext getApplicationContext() {
  14.         return applicationContext;
  15.     }
  16.  
  17.     //通过name获取 Bean.
  18.     public static Object getBean(String name) {
  19.         return getApplicationContext().getBean(name);
  20.     }
  21.  
  22.     //通过class获取Bean.
  23.     public static <T> T getBean(Class<T> clazz) {
  24.         return getApplicationContext().getBean(clazz);
  25.     }
  26.  
  27.     //通过name,以及Clazz返回指定的Bean
  28.     public static <T> T getBean(String name, Class<T> clazz) {
  29.         return getApplicationContext().getBean(name, clazz);
  30.     }
  31. }

这种做法实际上在flink流处理中也是不可行的，在我之前的flink文章中 Flink读写系列之-读mysql并写入mysql 其中读和写阶段有一个open方法，这个方法专门用于进行初始化的，那么我们可以在这里进行spring bean的初始化。那么MySink改造后即为：

  1. @EnableAutoConfiguration
  2. @MapperScan(basePackages = {"com.xxx.bigdata.xxx.mapper"})
  3. public class SimpleSink extends RichSinkFunction<String> {
  4.  
  5.  
  6.     TeacherInfoMapper teacherInfoMapper;
  7.  
  8.     @Override
  9.     public void open(Configuration parameters) throws Exception {
  10.         super.open(parameters);
  11.         SpringApplication application = new SpringApplication(SimpleSink.class);
  12.         application.setBannerMode(Banner.Mode.OFF);
  13.         ApplicationContext context = application.run(new String[]{});
  14.         teacherInfoMapper = context.getBean(TeacherInfoMapper.class);
  15.     }
  16.  
  17.     @Override
  18.     public void close() throws Exception {
  19.         super.close();
  20.     }
  21.  
  22.     @Override
  23.     public void invoke(String value, Context context) throws Exception {
  24.         List<TeacherInfo> teacherInfoList = teacherInfoMapper.selectByPage(0, 100);
  25.         teacherInfoList.stream().forEach(teacherInfo -> System.out.println("teacherinfo:" + teacherInfo.getTeacherId() + "," + teacherInfo.getTimeBit() + "," + teacherInfo.getWeek()));
  26.     }
  27. }

在invoke中就可以访问spring容器中的Mapper方法了。

 pom如下：

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  4.     <modelVersion>4.0.0</modelVersion>
  5.     <parent>
  6.         <groupId>org.springframework.boot</groupId>
  7.         <artifactId>spring-boot-starter-parent</artifactId>
  8.         <version>2.1.6.RELEASE</version>
  9.         <relativePath/> <!-- lookup parent from repository -->
  10.     </parent>
  11.     <groupId>com.xxx.bigdata</groupId>
  12.     <artifactId>flink-project</artifactId>
  13.     <version>1.0.0</version>
  14.     <name>flink-project</name>
  15.     <packaging>jar</packaging>
  16.     <description>My project for Spring Boot</description>
  17.  
  18.     <properties>
  19.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  20.         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
  21.         <java.version>1.8</java.version>
  22.         <flink.version>1.6.1</flink.version>
  23.         <skipTests>true</skipTests>
  24.         <maven.compiler.source>1.8</maven.compiler.source>
  25.         <maven.compiler.target>1.8</maven.compiler.target>
  26.     </properties>
  27.  
  28.     <dependencies>
  29.         <dependency>
  30.             <groupId>org.springframework.boot</groupId>
  31.             <artifactId>spring-boot-starter</artifactId>
  32.             <exclusions>
  33.                 <exclusion>
  34.                     <groupId>ch.qos.logback</groupId>
  35.                     <artifactId>logback-classic</artifactId>
  36.                 </exclusion>
  37.             </exclusions>
  38.         </dependency>
  39.         <dependency>
  40.             <groupId>org.apache.flink</groupId>
  41.             <artifactId>flink-java</artifactId>
  42.             <version>${flink.version}</version>
  43.         </dependency>
  44.         <dependency>
  45.             <groupId>org.apache.flink</groupId>
  46.             <artifactId>flink-streaming-java_2.11</artifactId>
  47.             <version>${flink.version}</version>
  48.         </dependency>
  49.         <dependency>
  50.             <groupId>org.apache.flink</groupId>
  51.             <artifactId>flink-connector-kafka-0.10_2.11</artifactId>
  52.             <version>${flink.version}</version>
  53.         </dependency>
  54.         <dependency>
  55.             <groupId>com.cloudera</groupId>
  56.             <artifactId>ImpalaJDBC41</artifactId>
  57.             <version>2.6.4</version>
  58.         </dependency>
  59.         <dependency>
  60.             <groupId>com.zaxxer</groupId>
  61.             <artifactId>HikariCP</artifactId>
  62.             <version>3.2.0</version>
  63.         </dependency>
  64.         <dependency>
  65.             <groupId>org.mybatis.spring.boot</groupId>
  66.             <artifactId>mybatis-spring-boot-starter</artifactId>
  67.             <version>1.3.1</version>
  68.         </dependency>
  69.         <dependency>
  70.             <groupId>com.alibaba</groupId>
  71.             <artifactId>fastjson</artifactId>
  72.             <version>1.2.47</version>
  73.         </dependency>
  74.         <dependency>
  75.             <groupId>org.projectlombok</groupId>
  76.             <artifactId>lombok</artifactId>
  77.             <optional>true</optional>
  78.         </dependency>
  79.  
  80.         <dependency>
  81.             <groupId>org.springframework.boot</groupId>
  82.             <artifactId>spring-boot-starter-test</artifactId>
  83.             <scope>test</scope>
  84.         </dependency>
  85.     </dependencies>
  86.  
  87.     <build>
  88.         <sourceDirectory>src/main/java</sourceDirectory>
  89.         <resources>
  90.             <resource>
  91.                 <directory>src/main/resources</directory>
  92.                 <filtering>true</filtering>
  93.                 <includes>
  94.                     <include>application.properties</include>
  95.                     <include>application-${package.environment}.properties</include>
  96.                 </includes>
  97.             </resource>
  98.         </resources>
  99.         <plugins>
  100.             <plugin>
  101.                 <groupId>org.springframework.boot</groupId>
  102.                 <artifactId>spring-boot-maven-plugin</artifactId>
  103.                 <configuration>
  104.                     <skip>true</skip>
  105.                     <mainClass>com.xxx.bigdata.xxx.Application</mainClass>
  106.                 </configuration>
  107.                 <executions>
  108.                     <execution>
  109.                         <goals>
  110.                             <goal>repackage</goal>
  111.                         </goals>
  112.                     </execution>
  113.                 </executions>
  114.             </plugin>
  115.             <!--mybatis plugin to generate mapping file and class-->
  116.             <plugin>
  117.                 <groupId>org.mybatis.generator</groupId>
  118.                 <artifactId>mybatis-generator-maven-plugin</artifactId>
  119.                 <version>1.3.5</version>
  120.                 <configuration>
  121.                     <configurationFile>${basedir}/src/main/resources/generatorConfig.xml</configurationFile>
  122.                     <overwrite>true</overwrite>
  123.                     <verbose>true</verbose>
  124.                 </configuration>
  125.                 <dependencies>
  126.                     <dependency>
  127.                         <groupId>com.cloudera</groupId>
  128.                         <artifactId>ImpalaJDBC41</artifactId>
  129.                         <version>2.6.4</version>
  130.                     </dependency>
  131.                 </dependencies>
  132.             </plugin>
  133.         </plugins>
  134.     </build>
  135.  
  136.     <profiles>
  137.         <!--开发环境-->
  138.         <profile>
  139.             <id>dev</id>
  140.             <properties>
  141.                 <package.environment>dev</package.environment>
  142.             </properties>
  143.             <!--默认环境-->
  144.             <activation>
  145.                 <activeByDefault>true</activeByDefault>
  146.             </activation>
  147.         </profile>
  148.         <!--预发布环境-->
  149.         <profile>
  150.             <id>pre</id>
  151.             <properties>
  152.                 <package.environment>pre</package.environment>
  153.             </properties>
  154.         </profile>
  155.         <!--生产环境-->
  156.         <profile>
  157.             <id>pro</id>
  158.             <properties>
  159.                 <package.environment>pro</package.environment>
  160.             </properties>
  161.         </profile>
  162.     </profiles>
  163.  
  164. </project>

项目打包使用了默认的spring boot插件，配置了skip为true，如果不配置此项，打包后会多一个BOOT-INF目录，运行时会引起ClassNotFoundException等各种异常，比如KafkaStreming问题，甚至需要反转flink的类加载机制，由child-first变为parent-first(修改flink配置文件)等等。

遇到的问题：

1.  java.lang.NoSuchMethodError: com.google.gson.GsonBuilder.setLenient()Lcom/google/gson/GsonBuilder

 GsonBuilder类来自gson-xxx.jar包，而我在自己的项目中执行mvn dependency:tree并没有发现依赖这个包。莫非在flink运行时会使用自己lib库下的gson包，转而去flink的lib库下，发现flink-dist_2.11-1.6.1.jar里包含了gson-xxx包，但是打开这个包一看类中没有setLenient方法，于是在服务器上建立一个commlib，把gson-2.8.0.jar（包含setLenient方法）放进去，然后使用flink run提交时，指定classpath即可。

  2.日志冲突

Caused by: java.lang.IllegalArgumentException: LoggerFactory is not a Logback LoggerContext but Logback is on the classpath. Either remove Logback or the competing implementation (class org.slf4j.impl.Log4jLoggerFactory loaded from file:/opt/flink-1.6.1/lib/slf4j-log4j12-1.7.7.jar). If you are using WebLogic you will need to add 'org.slf4j' to prefer-application-packages in WEB-INF/weblogic.xml: org.slf4j.impl.Log4jLoggerFactory

排除springboot中的日志即可：

  1. <dependency>
  2.        <groupId>org.springframework.boot</groupId>
  3.        <artifactId>spring-boot-starter</artifactId>
  4.        <exclusions>
  5.              <exclusion>
  6.                  <groupId>ch.qos.logback</groupId>
  7.                  <artifactId>logback-classic</artifactId>
  8.              </exclusion>
  9.        </exclusions>
  10. </dependency>

3.flink run提交作业到yarn上时，如果需要指定classpath，则需要指定到确定的jar包，指定目录不可行。那么假如所有依赖包已经放置在目录中，拼接的shell可以这么写：

  1.  
  2. lib_classpath="";
  3.  
  4. for jar in `ls /home/hadoop/lib`
  5. do
  6.   jar_suffix=${jar##*.}
  7.   if [ "$jar_suffix" = "jar" ]
  8.   then
  9.     jar_path=" --classpath file:///home/hadoop/lib/$jar "
  10.     lib_classpath=${lib_classpath}${jar_path}
  11.   else
  12.     echo "the jar file $jar it not legal jar file,skip appendig"
  13.   fi
  14. done

拼接后的lib_classpath值如下效果：

--classpath file:///home/hadoop/lib/accessors-smart-1.2.jar  --classpath file:///home/hadoop/lib/akka-actor_2.11-2.4.20.jar

注意：如果jar包放本地文件系统，那么需要每台机器都放一份。

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

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

  

闽ICP备14008679号