entrypoint for executing job in task manager

[Steven Wu]

Here is my understanding of how job submission works in Flink. When submitting a job to job manager via REST API, we provide a entry class. Job manager then evaluate job graph and ship serialized operators to task manager. Task manager then open operators and run tasks.

My app would typically requires some initialization phase to setup my own running context in task manager (e.g. calling a static method of some class). Does Flink provide any entry hook in task manager when executing a job (and tasks)? As for job manager, the entry class provides such hook where I can initialize my static context.

Thanks,

Steven

[Piotr Nowojski]

Open method is called just before any elements are processed. You can hook in any initialisation logic there, including initialisation of a static context. However keep in mind, that since this context is static, it will be shared between multiple operators (if you are running parallelism > number of task managers), so accesses to it must be synchronized (including initialisation). Another thing to consider is that managing the life cycle of static context can be tricky (when to close it and release it’s resources).

The questions is, whether you really need a static context?

Thanks,
Piotrek


> On 21 Dec 2017, at 07:53, Steven Wu <[hidden email]> wrote:
>
> Here is my understanding of how job submission works in Flink. When submitting a job to job manager via REST API, we provide a entry class. Job manager then evaluate job graph and ship serialized operators to task manager. Task manager then open operators and run tasks.
>
> My app would typically requires some initialization phase to setup my own running context in task manager (e.g. calling a static method of some class). Does Flink provide any entry hook in task manager when executing a job (and tasks)? As for job manager, the entry class provides such hook where I can initialize my static context.
>
> Thanks,
> Steven

[Steven Wu]

We use Guice for dependency injection. We need to install additional Guice modules (for bindings) when setting up this static context of Guice injector.

Calling the static initializer from operator open method won't really help. Not all operators are implemented by app developer who want to install additional Guice modules. E.g. kafka source operator is implemented/provided by our platform. I think the source operator will open first, which means app operator won't get a chance to initialize the static context. What would really help if there is a entry hook (at task manager) that is executed before any operator opening.

On Thu, Dec 21, 2017 at 12:27 AM, Piotr Nowojski <[hidden email]> wrote:

Open method is called just before any elements are processed. You can hook in any initialisation logic there, including initialisation of a static context. However keep in mind, that since this context is static, it will be shared between multiple operators (if you are running parallelism > number of task managers), so accesses to it must be synchronized (including initialisation). Another thing to consider is that managing the life cycle of static context can be tricky (when to close it and release it’s resources).

The questions is, whether you really need a static context?

Thanks,
Piotrek

> On 21 Dec 2017, at 07:53, Steven Wu <[hidden email]> wrote:
>
> Here is my understanding of how job submission works in Flink. When submitting a job to job manager via REST API, we provide a entry class. Job manager then evaluate job graph and ship serialized operators to task manager. Task manager then open operators and run tasks.
>
> My app would typically requires some initialization phase to setup my own running context in task manager (e.g. calling a static method of some class). Does Flink provide any entry hook in task manager when executing a job (and tasks)? As for job manager, the entry class provides such hook where I can initialize my static context.
>
> Thanks,
> Steven

[Piotr Nowojski]

I don’t think there is such hook in the Flink code now. You will have to walk around this issue somehow in user space. 

Maybe you could make a contract that every operator before touching Guice, should call static synchronized method `initializeGuiceContext`. This method could search the classpath for classes with some specific annotations, for example `@MyInitializationHook` and install/add all of such hooks before actually using Guice?

Piotrek

On 21 Dec 2017, at 17:49, Steven Wu <[hidden email]> wrote:

We use Guice for dependency injection. We need to install additional Guice modules (for bindings) when setting up this static context of Guice injector.

Calling the static initializer from operator open method won't really help. Not all operators are implemented by app developer who want to install additional Guice modules. E.g. kafka source operator is implemented/provided by our platform. I think the source operator will open first, which means app operator won't get a chance to initialize the static context. What would really help if there is a entry hook (at task manager) that is executed before any operator opening.

On Thu, Dec 21, 2017 at 12:27 AM, Piotr Nowojski <[hidden email]> wrote:

Open method is called just before any elements are processed. You can hook in any initialisation logic there, including initialisation of a static context. However keep in mind, that since this context is static, it will be shared between multiple operators (if you are running parallelism > number of task managers), so accesses to it must be synchronized (including initialisation). Another thing to consider is that managing the life cycle of static context can be tricky (when to close it and release it’s resources).

The questions is, whether you really need a static context?

Thanks,
Piotrek

> On 21 Dec 2017, at 07:53, Steven Wu <[hidden email]> wrote:
>
> Here is my understanding of how job submission works in Flink. When submitting a job to job manager via REST API, we provide a entry class. Job manager then evaluate job graph and ship serialized operators to task manager. Task manager then open operators and run tasks.
>
> My app would typically requires some initialization phase to setup my own running context in task manager (e.g. calling a static method of some class). Does Flink provide any entry hook in task manager when executing a job (and tasks)? As for job manager, the entry class provides such hook where I can initialize my static context.
>
> Thanks,
> Steven

[Stephan Ewen]

It would be great to understand a bit more what the exact requirements here are, and what setup you use.

I am not a dependency injection expert, so let me know if what I am suggesting here is complete bogus.

(1) Fix set of libraries for Dependency Injection, or dedicated container images per application

If you have a dedicated JM and TM Flink image that you build per job, I would assume that you also put all the required the libraries directly into the lib folder, so everything is available on startup.

In that case, could you just warp the TM and JM main methods to first call the initialization methods to set up dependency injection?

This would also work if you have container images that are not job-specific, but all the libraries relevant to dependency injection are part of the image (the lib folder).

(2) Generic container images, plus dynamic set of libraries for dependency injection

Assuming you do not have job-specific container images, and each application brings its own dependencies it wants to set up for dependency injection,

we could look in the following direction.

The dependencies need to be set up for each Task on the TaskManager  , because each task gets potentially a dedicated classloader.

Have you tried an approach like the following?

  - Create a static dependency initializer utility class that has a static " installModulesIfNotYetInstalled ()" method.

  - Each class that you use should have as the first line a static initializer block that calls that utility:

    public class MyFunction implements MapFunction<A, B> {

        static {

            InjectionUtil.installModulesIfNotYetInstalled();

        }

        public A map(B value) {...}

        ...

    }

  - You can probably create yourself a base class that does that from which all you functions extend.

On Fri, Dec 22, 2017 at 11:23 AM, Piotr Nowojski <[hidden email]> wrote:

I don’t think there is such hook in the Flink code now. You will have to walk around this issue somehow in user space. 

Maybe you could make a contract that every operator before touching Guice, should call static synchronized method `initializeGuiceContext`. This method could search the classpath for classes with some specific annotations, for example `@MyInitializationHook` and install/add all of such hooks before actually using Guice?

Piotrek

On 21 Dec 2017, at 17:49, Steven Wu <[hidden email]> wrote:

We use Guice for dependency injection. We need to install additional Guice modules (for bindings) when setting up this static context of Guice injector.

Calling the static initializer from operator open method won't really help. Not all operators are implemented by app developer who want to install additional Guice modules. E.g. kafka source operator is implemented/provided by our platform. I think the source operator will open first, which means app operator won't get a chance to initialize the static context. What would really help if there is a entry hook (at task manager) that is executed before any operator opening.

On Thu, Dec 21, 2017 at 12:27 AM, Piotr Nowojski <[hidden email]> wrote:

Open method is called just before any elements are processed. You can hook in any initialisation logic there, including initialisation of a static context. However keep in mind, that since this context is static, it will be shared between multiple operators (if you are running parallelism > number of task managers), so accesses to it must be synchronized (including initialisation). Another thing to consider is that managing the life cycle of static context can be tricky (when to close it and release it’s resources).

The questions is, whether you really need a static context?

Thanks,
Piotrek

> On 21 Dec 2017, at 07:53, Steven Wu <[hidden email]> wrote:
>
> Here is my understanding of how job submission works in Flink. When submitting a job to job manager via REST API, we provide a entry class. Job manager then evaluate job graph and ship serialized operators to task manager. Task manager then open operators and run tasks.
>
> My app would typically requires some initialization phase to setup my own running context in task manager (e.g. calling a static method of some class). Does Flink provide any entry hook in task manager when executing a job (and tasks)? As for job manager, the entry class provides such hook where I can initialize my static context.
>
> Thanks,
> Steven

[Steven Wu]

Thanks, let me clarify the requirement. Sorry that it wasn't clear in the original email.

Here is our setup. 

these 3 dirs are added to classpath

* flink/lib: core flink jars (like flink-dist_2.11, flink-shaded-hadoop2-uber)

* spaaslib: many jars pulled in our internal platform

* jobs: a single fat jar for app/job code lives here

running Flink

* "jobmanager.web.upload.dir" is configured to use the "jobs" dir above.

* we use REST api to submit job in a standalone cluster.

Here are the requirements for two level of init hooks

1) provide JVM init hook in JobManager and TaskManager class during JVM startup for user to extend. right now, we overrides the main method and others (as you were also suggesting). that is a little fragile as it is tight coupling. JobManager and TaskManager class don't seem to implemented for override. we have seen breaking changes when upgrading Flink from 1.2 -> 1.3 -> 1.4

2) job init hook during job execution. jobmanager computes the job graph, ship it to taskmanager, which then execute the job (open/run operators). 

  - my original email is to allow user (app developer) to supply additional Guice binding modules at taskmanager with a job init hook. then we can create a child injector with these additional modules. but Guice child injector has some issue with de-dup modules and run into duplicate binding problem. so we decided to not pursue this route

  - my colleague (Monal) mentioned another use case where we can leverage such job init hook at taskmanager. e.g. Inside job init hook, we can decide and attach EBS volumes based on key group assignment. Our understanding is that such key group assignment is calculated by jobmanager during job planning.

The problem with your suggestion of "InjectionUtil.installModulesIfNotYetInstalled()" is that it will also be loaded/executed inside JobManager when it loads this class. It is just a minor issue though.

Thanks,

Steven

On Wed, Mar 21, 2018 at 9:47 AM, Stephan Ewen <[hidden email]> wrote:

It would be great to understand a bit more what the exact requirements here are, and what setup you use.

I am not a dependency injection expert, so let me know if what I am suggesting here is complete bogus.

(1) Fix set of libraries for Dependency Injection, or dedicated container images per application

If you have a dedicated JM and TM Flink image that you build per job, I would assume that you also put all the required the libraries directly into the lib folder, so everything is available on startup.

In that case, could you just warp the TM and JM main methods to first call the initialization methods to set up dependency injection?

This would also work if you have container images that are not job-specific, but all the libraries relevant to dependency injection are part of the image (the lib folder).

(2) Generic container images, plus dynamic set of libraries for dependency injection

Assuming you do not have job-specific container images, and each application brings its own dependencies it wants to set up for dependency injection,

we could look in the following direction.

The dependencies need to be set up for each Task on the TaskManager  , because each task gets potentially a dedicated classloader.

Have you tried an approach like the following?

  - Create a static dependency initializer utility class that has a static " installModulesIfNotYetInstalled ()" method.

  - Each class that you use should have as the first line a static initializer block that calls that utility:

    public class MyFunction implements MapFunction<A, B> {

        static {

            InjectionUtil.installModulesIfNotYetInstalled();

        }

        public A map(B value) {...}

        ...

    }

  - You can probably create yourself a base class that does that from which all you functions extend.

On Fri, Dec 22, 2017 at 11:23 AM, Piotr Nowojski <[hidden email]> wrote:

I don’t think there is such hook in the Flink code now. You will have to walk around this issue somehow in user space. 

Maybe you could make a contract that every operator before touching Guice, should call static synchronized method `initializeGuiceContext`. This method could search the classpath for classes with some specific annotations, for example `@MyInitializationHook` and install/add all of such hooks before actually using Guice?

Piotrek

On 21 Dec 2017, at 17:49, Steven Wu <[hidden email]> wrote:

We use Guice for dependency injection. We need to install additional Guice modules (for bindings) when setting up this static context of Guice injector.

Calling the static initializer from operator open method won't really help. Not all operators are implemented by app developer who want to install additional Guice modules. E.g. kafka source operator is implemented/provided by our platform. I think the source operator will open first, which means app operator won't get a chance to initialize the static context. What would really help if there is a entry hook (at task manager) that is executed before any operator opening.

On Thu, Dec 21, 2017 at 12:27 AM, Piotr Nowojski <[hidden email]> wrote:

Open method is called just before any elements are processed. You can hook in any initialisation logic there, including initialisation of a static context. However keep in mind, that since this context is static, it will be shared between multiple operators (if you are running parallelism > number of task managers), so accesses to it must be synchronized (including initialisation). Another thing to consider is that managing the life cycle of static context can be tricky (when to close it and release it’s resources).

The questions is, whether you really need a static context?

Thanks,
Piotrek

> On 21 Dec 2017, at 07:53, Steven Wu <[hidden email]> wrote:
>
> Here is my understanding of how job submission works in Flink. When submitting a job to job manager via REST API, we provide a entry class. Job manager then evaluate job graph and ship serialized operators to task manager. Task manager then open operators and run tasks.
>
> My app would typically requires some initialization phase to setup my own running context in task manager (e.g. calling a static method of some class). Does Flink provide any entry hook in task manager when executing a job (and tasks)? As for job manager, the entry class provides such hook where I can initialize my static context.
>
> Thanks,
> Steven

[Jimmy (Yi Pin) Cao]

Well this thread was super helpful. I'm also looking for ways to integrate DI

In particular something like the JVM init hooks for TM and JM would be nice. 

On Wed, Mar 21, 2018 at 5:08 PM, Steven Wu <[hidden email]> wrote:

Thanks, let me clarify the requirement. Sorry that it wasn't clear in the original email.

Here is our setup. 

these 3 dirs are added to classpath

* flink/lib: core flink jars (like flink-dist_2.11, flink-shaded-hadoop2-uber)

* spaaslib: many jars pulled in our internal platform

* jobs: a single fat jar for app/job code lives here

running Flink

* "jobmanager.web.upload.dir" is configured to use the "jobs" dir above.

* we use REST api to submit job in a standalone cluster.

Here are the requirements for two level of init hooks

1) provide JVM init hook in JobManager and TaskManager class during JVM startup for user to extend. right now, we overrides the main method and others (as you were also suggesting). that is a little fragile as it is tight coupling. JobManager and TaskManager class don't seem to implemented for override. we have seen breaking changes when upgrading Flink from 1.2 -> 1.3 -> 1.4

2) job init hook during job execution. jobmanager computes the job graph, ship it to taskmanager, which then execute the job (open/run operators). 

  - my original email is to allow user (app developer) to supply additional Guice binding modules at taskmanager with a job init hook. then we can create a child injector with these additional modules. but Guice child injector has some issue with de-dup modules and run into duplicate binding problem. so we decided to not pursue this route

  - my colleague (Monal) mentioned another use case where we can leverage such job init hook at taskmanager. e.g. Inside job init hook, we can decide and attach EBS volumes based on key group assignment. Our understanding is that such key group assignment is calculated by jobmanager during job planning.

The problem with your suggestion of "InjectionUtil.installModulesIfNotYetInstalled()" is that it will also be loaded/executed inside JobManager when it loads this class. It is just a minor issue though.

Thanks,

Steven

On Wed, Mar 21, 2018 at 9:47 AM, Stephan Ewen <[hidden email]> wrote:

It would be great to understand a bit more what the exact requirements here are, and what setup you use.

I am not a dependency injection expert, so let me know if what I am suggesting here is complete bogus.

(1) Fix set of libraries for Dependency Injection, or dedicated container images per application

If you have a dedicated JM and TM Flink image that you build per job, I would assume that you also put all the required the libraries directly into the lib folder, so everything is available on startup.

In that case, could you just warp the TM and JM main methods to first call the initialization methods to set up dependency injection?

This would also work if you have container images that are not job-specific, but all the libraries relevant to dependency injection are part of the image (the lib folder).

(2) Generic container images, plus dynamic set of libraries for dependency injection

Assuming you do not have job-specific container images, and each application brings its own dependencies it wants to set up for dependency injection,

we could look in the following direction.

The dependencies need to be set up for each Task on the TaskManager  , because each task gets potentially a dedicated classloader.

Have you tried an approach like the following?

  - Create a static dependency initializer utility class that has a static " installModulesIfNotYetInstalled ()" method.

  - Each class that you use should have as the first line a static initializer block that calls that utility:

    public class MyFunction implements MapFunction<A, B> {

        static {

            InjectionUtil.installModulesIfNotYetInstalled();

        }

        public A map(B value) {...}

        ...

    }

  - You can probably create yourself a base class that does that from which all you functions extend.

On Fri, Dec 22, 2017 at 11:23 AM, Piotr Nowojski <[hidden email]> wrote:

I don’t think there is such hook in the Flink code now. You will have to walk around this issue somehow in user space. 

Maybe you could make a contract that every operator before touching Guice, should call static synchronized method `initializeGuiceContext`. This method could search the classpath for classes with some specific annotations, for example `@MyInitializationHook` and install/add all of such hooks before actually using Guice?

Piotrek

On 21 Dec 2017, at 17:49, Steven Wu <[hidden email]> wrote:

We use Guice for dependency injection. We need to install additional Guice modules (for bindings) when setting up this static context of Guice injector.

Calling the static initializer from operator open method won't really help. Not all operators are implemented by app developer who want to install additional Guice modules. E.g. kafka source operator is implemented/provided by our platform. I think the source operator will open first, which means app operator won't get a chance to initialize the static context. What would really help if there is a entry hook (at task manager) that is executed before any operator opening.

On Thu, Dec 21, 2017 at 12:27 AM, Piotr Nowojski <[hidden email]> wrote:

Open method is called just before any elements are processed. You can hook in any initialisation logic there, including initialisation of a static context. However keep in mind, that since this context is static, it will be shared between multiple operators (if you are running parallelism > number of task managers), so accesses to it must be synchronized (including initialisation). Another thing to consider is that managing the life cycle of static context can be tricky (when to close it and release it’s resources).

The questions is, whether you really need a static context?

Thanks,
Piotrek

> On 21 Dec 2017, at 07:53, Steven Wu <[hidden email]> wrote:
>
> Here is my understanding of how job submission works in Flink. When submitting a job to job manager via REST API, we provide a entry class. Job manager then evaluate job graph and ship serialized operators to task manager. Task manager then open operators and run tasks.
>
> My app would typically requires some initialization phase to setup my own running context in task manager (e.g. calling a static method of some class). Does Flink provide any entry hook in task manager when executing a job (and tasks)? As for job manager, the entry class provides such hook where I can initialize my static context.
>
> Thanks,
> Steven

[Bob Tiernay]

Hi Steven,

Curious how you solved this for you use case. Did you ever find a
satisfactory approach?

Thanks in advance,

Bob



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/

[Bob Tiernay]

Please see  FLINK-14184 <https://issues.apache.org/jira/browse/FLINK-14184>  
which should fully support such use cases in the future. Feel free to vote
for it if you believe it would help your use case.



--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/