Support for various job submission types and job resource types about flink-on-k8s-operator HOT 6 OPEN

Thanks for the great feature requests!

In terms of job resource staging,we definitely want to support remote file (see #15). Currently, job (client) image is the same as jobmanager / taskmanager image, because we use Flink CLI to submit job. Internally, I created a custom image (not open-source) based on the Flink image and added the capability of downloading remote job file if the job jar URL is http(s)://, gs://... In the controller, we use an env variable FLINK_JOB_JAR_URI to pass the job jar URI to the job client container, the job client container first downloads it to local filesystem, then submit the job to Flink.

Now, I'm thinking about use the operator image as job (client) image, add the feature in the operator image. This simplifies the deployment, users only need an operator image + the unmodified official Flink image.

In terms of job types, I need more investigation about Apache Beam Python.

Your ideas and contributions are highly appreciated, thank you!

from flink-on-k8s-operator.

It would be great if these features were introduced. Thank you for your interest and work on this feature request!

from flink-on-k8s-operator.

We added init containers for job Pod in #100, it can be used to download remote job jar and other artifacts, for example. Do you think it is good enough for remote job jar?

I just tested Beam Python on Flink Operator these days, there are still some gaps in Beam, e.g. BEAM-8660. We might need to wait for Beam 2.18. There are also other improvements planned in Beam, then I think it should not be too hard to support Beam Python on Flink Operator.

from flink-on-k8s-operator.

It seems to be a good way to deal with various remote resources. Thanks!

With regard to Beam, let's take look at the related issues.

from flink-on-k8s-operator.

IMO, the operator could support the Apache beam job in two approaches. Below are two ways I've thought. More generalized implementation would be preferable if possible, but it looks like it's hard to take advantage of existing fields like fromSavepoint or parallelism and so on.

1. Generalized implementation

job:

  command: python

  args: ['-m', 'apache_beam.examples.wordcount', '--job_endpoint=localhost:8099', '--environment_type=EXTERNAL', '--environment_config=localhost:50000']

  sidecars:

  - name: beam-job-server

    image: apachebeam/flink1.9_job_server
    args: ["--flink-master-url=jobmanager:8081"]

• If you add a custom command, args and sidecars - to deploy pod-local beam job server - to the job spec, beam job execution becomes possible

• Users should add the arguments related to job submission including job server endpoint and parallelism to args. Only job submission is possible and fields of job spec / status related to Flink would be useless.

2. Implementation that supports Apache beam

<CR example 1>

job:

  type: beamPython
  command: python
  args: ['-m', 'apache_beam.examples.wordcount']

<CR example 2>

job:

  type: beamPython

  command: python

  appPath: /cache/examples

  entrypoint: apache_beam.examples.wordcount

• Add type to job spec and implement logics of job submissions and savepointing by each type. Sidecar job server container would be appended by operator.
  job type ex: flink java, flink python, beam java, beam python, custom

If we support multiple job types, we need to add python program to args or change the jarFile and className of the jobSpec to a more general representation.
ex. jarFile-> appPath, className-> entrypoint

Do you have any implementation plans or ideas? If so, which method is it close to?
I need this feature first, so I wonder what your plan is. :)

from flink-on-k8s-operator.

If there is no implementation plan in the near future I would like to contribute beam feature.

from flink-on-k8s-operator.