I am using pyflink with k8s operator and i want to create a application with multiple jobs. How can i do that with pyflink ?

What do you mean by

an application with multiple jobs

?

I am pretty new and trying to learn. I mean, I deployed one deployment and i wrote “job” parameter inside of “spec”. But it is deploying only one job. I have 2 different python files and i want to deploy 2 different job inside of 1 deployment

Could you check if https://github.com/apache/flink-kubernetes-operator/blob/main/examples/basic-session-deployment-and-job.yaml could meets your requirements.

Yes i saw this, but i couldn’t configured for python. To be more specific i am trying to configure like this;

apiVersion: <http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>
kind: FlinkDeployment
metadata:
  name: test
  namespace: flink
spec:
  ingress:
    template:  "{{name}}.<http://test.com|test.com>"
    className: "nginx"
    annotations: {}
  image: <priv pyflink repo>
  imagePullPolicy: "Always"
  flinkVersion: v1_16
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "1"
  serviceAccount: flink
  jobManager:
    podTemplate:
      apiVersion: v1
      kind: Pod
      metadata:
       name: pod-template
      spec:
        serviceAccount: flink
        containers:
        # Do not change the main container name
          - name: flink-main-container
            envFrom:
            - secretRef:
                name: flink-secret
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "2048m"
      cpu: 1
    replicas: 1

---

apiVersion: <http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>
kind: FlinkSessionJob
metadata:
  name: test1
  namespace: flink
spec:
  deploymentName: test
  job:
    jarURI: file:///opt/flink/opt/flink-python_2.12-1.16.1.jar # Note, this jarURI is actually a placeholder
    entryClass: "org.apache.flink.client.python.PythonDriver"
    args: ["-pyclientexec", "/usr/file/bin/python3", "-py", "/opt/flink/usrlib/test1.py"]
    parallelism: 4
    upgradeMode: stateless
    
---
apiVersion: <http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>
kind: FlinkSessionJob
metadata:
  name: test2
  namespace: flink
spec:
  deploymentName: test
  job:
    jarURI: file:///opt/flink/opt/flink-python_2.12-1.16.1.jar # Note, this jarURI is actually a placeholder
    entryClass: "org.apache.flink.client.python.PythonDriver"
    args: ["-pyclientexec", "/usr/file/bin/python3", "-py", "/opt/flink/usrlib/test2.py"]
    parallelism: 4
    upgradeMode: stateless

But i am getting “org.apache.flink.kubernetes.operator.exception.ReconciliationException: java.io.FileNotFoundException: /opt/flink/opt/flink-python_2.12-1.16.1.jar (No such file or directory)”

You could try putting the PythonDriver in your Flink Lib folder in your session cluster

@Gyula Fóra , Thanks for your response. I think the session job has been created from flink-k8s-operator pod. If i create custom image with python jar is it possible to work ?

@Mali I guess you could use

local:///opt/flink/opt/flink-python_2.12-1.16.1.jar

instead of

file:///opt/flink/opt/flink-python_2.12-1.16.1.jar

@Dian Fu I already tried it actually, but i got “No scheme exists for ‘local’ “. But “local:///” is working without issue with writing deployment only 🙂

Oh, the jar should be

flink-python-1.16.1.jar

. It has removed the

2.12

suffix since 1.16.

really ? let me try it. Very thanks @Dian Fu i will let you know

Just want to add some background: as the comment said, the jarURI is actually a placeholder to pass the check of the operator. So it is totally ok to use another jar existed in the image to do that.

@Dian Fu @Gyula Fóra @Biao Geng I tried but it is not exists i added manually but getting different jars errors. Maybe you can refer me to what should i do from scratch. I deployed flink-kubernetes-operator via helm and i created custom pyflink docker image. I created deployment yaml and my pyflink code is working successfully now. What should i do to append extra jobs. Actually, i didn’t create any Flink Session Cluster but when i tried to “basic-session-example.yaml” it is working well. Maybe i am doing mistake at the beginnings. Many thanks..

@Mali Could you share the current exception message and also the latest configuration?

@Dian Fu Hello, of course, i will send you. Just to understand clearly, Is it possible to submit multiple jobs using Application Mode ?

I think this is not expected and so the behavior may be undefined.

@Dian Fu I am getting; org.apache.flink.util.FlinkRuntimeException: Cannot have more than one execute() or executeAsync() call in a single environment. Actually according to documentation: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/overview/#application-mode It is saying it allows multi-execute() operations. According to this document, i should able to create multi jobs using multi execute() ? Am i miss something ?

Yes, you could do that. But it also states that there are some limitations in the documentation you shared:

"The Application Mode allows for multi-execute() applications but High-Availability is not supported in these cases. High-Availability in Application Mode is only supported for single-execute() applications.

Additionally, when any of multiple running jobs in Application Mode (submitted for example using executeAsync()) gets cancelled, all jobs will be stopped and the JobManager will shut down. Regular job completions (by the sources shutting down) are supported."

If these limitations are not much a problem for your cases, I think you could try it~

I tried it but it got an error sent above. It is interesting i closed HA also

How you disable the HA? Have you set

high-availability

to

None

? https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/config/#high-availability-1

The default value is “None” thats why i didn’t write anything but i will try it