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.

Usage notes

In the examples in this topic, the JAR file is packaged directly in the image. To use your own JAR file, upload it to OSS. For more information, see Simple upload.

If you use a JAR file from OSS, change local:///opt/flink/examples/streaming/StateMachineExample.jar to its path. The path format is oss://<yourBucketName>/<path>.jar.

Method 1: Use 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 from YAML in the upper-right corner.

4. On the create page, select Custom from the Sample Template drop-down 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 Flink version other than 1.13, update the flinkVersion parameter to the version shown in the console.

Method 2: Use kubectl

1. Connect to the Kubernetes cluster with kubectl. For more information, see Obtain the kubeconfig of an ACK cluster and use kubectl to connect to the cluster.

   You can also connect to the Kubernetes cluster by 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 customize the filename. This example uses basic-emr-example.yaml.

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

3. Run the following command to submit the job.

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

   Note

   Replace <your_cluster_namespace> with the namespace of your cluster. You can find the namespace on the Cluster Details page in the ACK console.