Spring 与 Apache Beam

我想将 Spring 与 Apache Beam 结合使用，它将在 Google Cloud Data flow Runner 上运行。数据流作业应该能够在执行管道步骤时使用 Spring 运行时应用程序上下文。我想在 Apache Beam 管道中使用 Spring 功能来实现 DI 和其他功能。在 google 上浏览了几个小时后，我找不到任何显示 Apache Beam 中 Spring 集成的帖子或文档。所以，如果有人用 Apache Beam 尝试过 Spring，请告诉我。

在主类中，我已经初始化了 spring 应用程序上下文，但在执行管道步骤时它不可用。我收到自动装配 bean 的空指针异常。我想问题是，在运行时上下文对工作线程不可用。

 public static void main(String[] args) {
    initSpringApplicationContext();

    GcmOptions options = PipelineOptionsFactory.fromArgs(args)
        .withValidation()
        .as(GcmOptions.class);
    Pipeline pipeline = Pipeline.create(options);
    // pipeline definition
}

我想将 spring 应用程序上下文注入到每个 ParDo 函数中。

这里的问题是 ApplicationContext 在任何工作线程上都不可用，因为main方法仅在构造作业时调用，而不是在任何工作机器上调用。所以，initSpringApplicationContext从未召唤过任何工人。

我从未尝试过在 Apache Beam 中使用 Spring，但我想移动initSpringApplicationContext在静态初始化块中将导致您的预期结果。

public class ApplicationContextHolder {

    private static final ApplicationContext CTX;

    static {
        CTX = initApplicationContext();
    }

    public static ApplicationContext getContext() {
        return CTX;
    }
}

请注意，这不应被视为在 Apache Beam 中使用 Spring 的最佳实践，因为它不能很好地集成到 Apache Beam 的生命周期中。例如，当应用程序上下文初始化期间发生错误时，它会出现在应用程序上下文的第一个位置。ApplicationContextHolder用来。因此，我建议提取initApplicationContext脱离静态初始化程序块并根据 Apache Beam 的生命周期显式调用它。这setup https://beam.apache.org/releases/javadoc/2.13.0/org/apache/beam/sdk/transforms/DoFn.Setup.html阶段将是一个好地方。