Flink与Spring Boot的集成_springboot 集成 flink

• flink 1.10.0
• spring boot 2.2.2.RELEASE

这方面的资料很少，网上有的方法不完整。基本思路是把spring容器的初始化放在sink的open方法中执行。

要么只使用spring framework组件，甚至使用xml这样的方式配置bean（使用ClassPathXmlApplicationContext ）；要么直接在open中启动了SpringApplication。有可能在单机的flink上能运行，但是on yarn的时候不行了。

其实想要达到的目的很简单：

• 使用spring的IoC容器管理组件
• 注解配置和自动配置
• 使用springboot的外部化配置（application.yml）
• 需要可以运行在yarn集群中

但是里面的坑非常多，除了需要了解一些flink的任务提交部署原理，需要对spring framework， spring boot， maven，hadoop yarn有一些了解。有些地方需要深入了解，否则莫名其妙入坑，半天爬不出来。所以需要记录一下，已方便后来者别再浪费时间。Flink的官方文档真的是很简短。

Spring容器的集成点

需要算子（Operator）具体Function（Source、Sink）的初始化中，因为这些算子会被序列化到分布式计算节点中执行。所以通常的main只是任务提交的入口，并不是最终算子执行初始化入口。

所以，在Source和Sink的open方法中初始化容器。由于通常Source都由比较固定的组件，比如kafka集成了FlinkKafkaConsumer，所以这部分没有过多的需要编写处理逻辑，从而没有引入spring容器。但是，初始化这部分组件有配置参数传递的需求。

依赖jar包注意点

引入flink或者hadoop等等这种运行时会提供的jar包时，记得把作用域置scope设为provided。

典型的flink依赖包引入如下：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-java</artifactId>
  <version>1.10.0</version>
  <scope>provided</scope>
</dependency>
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-java_2.11</artifactId>
  <version>1.10.0</version>
  <scope>provided</scope>
</dependency>
1
2
3
4
5
6
7
8
9
10
11
12

如果是kafka连接器这样的依赖，非flink核心依赖，则是需要打包时打进去的，使用默认的scope就行。

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka-0.10_2.11</artifactId>
    <version>1.10.0</version>
</dependency>
1
2
3
4
5

这里是一个自定义Sink的例子，在open中初始化spring的容器。

@Slf4j
public 
1