Весна с Apache Beam

Question

Я хочу использовать Spring с Apache Beam, который будет работать в Google Cloud Data Runner. Задание потока данных должно иметь возможность использовать контекст приложения Spring Runtime при выполнении шагов конвейера. Я хочу использовать функцию Spring в своем конвейере Apache Beam для DI и других вещей. После нескольких часов просмотра в Google, я не смог найти ни одного поста или документации, которая показывает интеграцию Spring в Apache Beam. Итак, если кто-то пробовал пружину с Apache Beam, пожалуйста, дайте мне знать.

В основном классе я инициализировал контекст приложения 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.

Answer 1

Проблема здесь в том, что ApplicationContext недоступен ни для одного работника, так как метод main вызывается только при создании задания, а не на любом рабочем компьютере. Следовательно, initSpringApplicationContext никогда не вызывается ни для какого работника.

Я никогда не пытался использовать Spring в Apache Beam, но, думаю, перемещение initSpringApplicationContext в блок статического инициализатора приведет к ожидаемому результату.

public class ApplicationContextHolder {

    private static final ApplicationContext CTX;

    static {
        CTX = initApplicationContext();
    }

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

Имейте в виду, что это само по себе не должно рассматриваться как лучший способ использования Spring в Apache Beam, так как он плохо интегрируется в жизненный цикл Apache Beam. Например, когда ошибка происходит во время инициализации контекста приложения, она будет появляться в первую очередь, где используется ApplicationContextHolder. Поэтому я бы рекомендовал извлечь initApplicationContext из блока статического инициализатора и явно вызывать его в отношении жизненного цикла Apache Beam. Фаза setup была бы хорошим местом для этого.