How to pass java arguments to Flink job artifacts in Application Mode

Question

I just upgrade the Flink from version 1.10 to 1.11. In 1.11, Flink provides new features that users can deploy the job in Application Mode on Kubernetes. https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/deployment/kubernetes.html#deploy-session-cluster

In V1.10, we start the Flink K8s cluster and then submit the job to Flink by run

exec ./bin/flink run \
  -d \
  /streakerflink_deploy.jar \
    --arg1 blablabla
    --arg2 blablabla
    --arg3 blablabla
    ...

We pass the java arguments through this command.

But, in V1.11, if we run Application mode, we don't need to run the flink run command above. I am wondering how do we pass the arguments to Flink job in Application mode (aka Job Cluster)?

Any help will be appreciated!

Yang Wang Yang Wang · Accepted Answer · 2020-09-04T03:23:49

Since you are using helm chart to start the Flink cluster on Kubernetes(aka K8s), i assume you are talking about the standalone K8s mode. Actually, the Application mode is very similar to the job cluster in 1.10 and before. So you could set the job arguments in the args field in the jobmanager-job.yaml just like following.

...
args: ["standalone-job", "--job-classname", "org.apache.flink.streaming.examples.join.WindowJoin", "--windowSize", "3000", "--rate", "100"]
...

If you really mean the native K8s mode, then it could be directly added after the flink run-application command.

$ ./bin/flink run-application -p 8 -t kubernetes-application \
  -Dkubernetes.cluster-id=<ClusterId> \
  -Dtaskmanager.memory.process.size=4096m \
  -Dkubernetes.taskmanager.cpu=2 \
  -Dtaskmanager.numberOfTaskSlots=4 \
  -Dkubernetes.container.image=<CustomImageName> \
  local:///opt/flink/examples/streaming/WindowJoin.jar \
  --windowSize 3000 --rate 100

Note: Please keep in mind that the key difference between standalone K8s and native K8s mode is the dynamic resource allocation. In native mode, we have an embedded K8s client, so Flink JobManager could allocate/release TaskManager pods on demands. Currently, native mode could only be used in Flink commands(kubernetes-session.sh, flink run-application).

How to pass java arguments to Flink job artifacts in Application Mode

2 Answers