E-MapReduce:Submit a Flink job

Prerequisites

You have created a Flink cluster in the EMR on ACK console. For more information, see Getting started.

Notes

In the examples in this topic, the JAR file is already packaged in the image. If you use your own JAR file, upload it to Alibaba Cloud OSS. For more information, see Simple upload.

You must then replace local:///opt/flink/examples/streaming/StateMachineExample.jar in the configuration with the path to your JAR file in Alibaba Cloud OSS. The path must use the format oss://<yourBucketName>/<path>.jar.

Method 1: Submit a job from the ACK console

1. Log on to the EMR on ACK console.

2. On the EMR on ACK page, find the target cluster and click the link in the ACK Cluster column.

3. On the Pods page, click Create Resources in YAML in the upper-right corner.

4. On the create page, select Custom from the Sample Template list, paste the following content 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

   Note

   This example uses Flink 1.13. If you use a different Flink version, update the flinkVersion parameter accordingly.

Method 2: Submit a job using kubectl

1. Connect to your Kubernetes cluster using kubectl. For more information, see Obtain the KubeConfig file of a cluster and use kubectl to connect to the cluster.

   You can also connect to the Kubernetes cluster using other methods, such as an API. For more information, see Use the Kubernetes API.

2. 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

   Note

   • You can use a custom file name. This example uses basic-emr-example.yaml.

   • This example uses Flink 1.13. If you use a different Flink version, update the flinkVersion parameter accordingly.

3. Run the following command to submit the job.

   kubectl apply -f basic-emr-example.yaml --namespace <cluster-namespace>

   Note

   Replace <cluster-namespace> with your cluster's namespace. You can find the namespace on the cluster management page in the EMR on ACK console.