Apache Beam Initializer

发布于 2025-01-18 09:48:51 字数 978 浏览 4 评论 0 原文

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

我已经将配置工厂初始化代码 +审核记录放在父级 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：org.devoteam.customjob

in platformInitialializer，我有一个Initilize（）方法（）包含配置工厂的初始化逻辑，还可以记录一些初始审核消息。

public class PlatformInitializer {

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

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

原文

In my dataflow job, I need to initialize a Config factory and log certain messages in an audit log before actual processing begins.

I have placed the Config factory initialization code + audit logging in a parent class PlatformInitializer and extending that in my Main Pipeline class.

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);
    }

as a result, I had to also implement Serializable interface in my Pipeline class because beam was throwing error - java.io.NotSerializableException: org.devoteam.CustomJob

Inside PlatformInitializer, I have an initilize() method that contains initialization logic for config factory and also log some initial audit messages.

public class PlatformInitializer {

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

My question is - is this right way to invoke some code that needs to be called before pipeline begins execution?

分享到QQ

分享到微博