如何在flink环境中初始化flink作业的spring资源

Question

最近在开发flink作业时遇到了一些问题，引入了spring和hibernate，作业会在flink集群上运行。所以我需要在任务管理器而不是作业管理器上运行 flink 操作符之前初始化 spring 资源。但是我找不到任何合适的 StreamExecutionEnvironment 方法来做到这一点。

我已经尝试了以下一些方法：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
ApplicationContext context = new ClassPathXmlApplicationContext("classpath:applicationContext.xml");
// etl business logic as flink operators
FlinkOperators.run();  
env.execute();

但是，当并行度多于一个的flink作业执行时，sp​​ring初始化将不在在每个task manager进程中。所以我不能在 flink 工作中使用 spring。

有什么方法可以在 flink 作业上初始化 spring 资源吗？

谢谢。

最好的问候。

阿尔文

Answer 1

每当您需要为每个任务管理器进行某种上下文初始化时，构建一个静态函数（如果您使用 scala，则为对象内部的函数）将这些初始化值存储在静态变量中。

这应该足够了，因为静态值存储在每个任务管理器的内存中。

我使用这种方法在每个任务管理器中加载属性文件，这些属性文件包含每个作业的配置。如果您正在加载文件，请检查每个任务管理器是否都有您要加载的文件的副本。