Submit a Flink job on EMR on ACK by using the ACK console or kubectl - E-MapReduce

Submit a Flink job to an EMR on ACK cluster using the ACK console or kubectl.

Prerequisites

Before you begin, make sure that you have:

A Flink cluster created on the EMR on ACK page of the Alibaba Cloud E-MapReduce (EMR) console. For setup instructions, see Getting started.

Method 1: Use the ACK console

Log on to the EMR on ACK console.
On the EMR on ACK page, find the target EMR cluster and click the link in the ACK Cluster column.
In the upper-right corner of the Pods page, click Create from YAML.

On the Create page, select Custom from the Sample Template drop-down list, paste the following YAML into the editor, and then click Create.

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: basic-emr-example
spec:
  flinkVersion: v1_13
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
    state.savepoints.dir: file:///flink-data/flink-savepoints
    state.checkpoints.dir: file:///flink-data/flink-checkpoints
  serviceAccount: flink
  podTemplate:
    spec:
      serviceAccount: flink
      containers:
        - name: flink-main-container
          volumeMounts:
            - mountPath: /flink-data
              name: flink-volume
      volumes:
        - name: flink-volume
          emptyDir: {}

  jobManager:
    replicas: 1
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "2048m"
      cpu: 1

  job:
    jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
    parallelism: 2
    upgradeMode: stateless

This example uses Flink 1.13. If you use a different Flink version, set flinkVersion to the version shown in the EMR console.

Method 2: Use kubectl

Prerequisites

Before you begin, make sure that you have:

kubectl installed on your local machine
A kubeconfig file configured to connect to the ACK cluster. For setup instructions, see Obtain the kubeconfig file of a cluster and use kubectl to connect to the cluster. Alternatively, connect by calling an API operation — see Use the Kubernetes API.

Steps

Create a file named basic-emr-example.yaml with the following content:

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: basic-emr-example
spec:
  flinkVersion: v1_13
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
    state.savepoints.dir: file:///flink-data/flink-savepoints
    state.checkpoints.dir: file:///flink-data/flink-checkpoints
  serviceAccount: flink
  podTemplate:
    spec:
      serviceAccount: flink
      containers:
        - name: flink-main-container
          volumeMounts:
            - mountPath: /flink-data
              name: flink-volume
      volumes:
        - name: flink-volume
          emptyDir: {}

  jobManager:
    replicas: 1
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "2048m"
      cpu: 1

  job:
    jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
    parallelism: 2
    upgradeMode: stateless

This example uses Flink 1.13. If you use a different version, update flinkVersion accordingly. The filename can be changed to suit your project conventions.

Submit the Flink job:
```
kubectl apply -f basic-emr-example.yaml -namespace <namespace>
```
Replace <namespace> with the namespace where your cluster resides. To find the namespace, log on to the EMR console and go to the Cluster Details tab.

What's next

View the logs of a Flink job and access the web UI of the job