Apache波束初始化器

Question

在我的数据流作业中，我需要初始化配置工厂，并在实际处理开始之前将某些消息记录在审核日志中。

我将配置工厂初始化代码审计日志记录放在父类PlatformInitializer中，并在我的主管道类中扩展它。

public class CustomJob extends PlatformInitializer implements Serializable {
    public static void main(String[] args) throws PropertyVetoException {
        CustomJob myCustomjob = new CustomJob();

        // Initialize config factories
        myCustomjob.initialize();

        // trigger dataflow job
        myCustomjob.parallelRead(args);
    }

因此，我还必须在我的管道类中实现可序列化接口，因为beam抛出了错误-<代码>java。io。NotSerializableException：组织。德维塔姆。自定义作业

在PlatformInitializer中，我有一个initilize（）方法，该方法包含配置工厂的初始化逻辑，并记录一些初始审核消息。

public class PlatformInitializer {

    public void initialize() {
        // Configfactory factory = new Configfactory()
        // CustomLogger.log("JOB-BEGIN-EVENT" + Current timestamp )
    }
}

我的问题是——这是调用某些需要在管道开始执行之前调用的代码的正确方法吗？

Answer 1

如果在运行时（而不是在管道构建时）需要初始化对象，则应将初始化逻辑移动到Beam DoFn<代码>DoFn有许多方法注释，可用于表示应在不同生命周期阶段执行的方法<代码>设置和开始绑定注释可能对您的用例有用。有关更多详细信息，请参见此处。