kubectl apply -f cert-manager.yaml

helm install -f values.yaml flink-kubernetes-operator . --namespace flink --create-namespace

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment  # Flink集群在K8s的资源类型
metadata:
  name: basic-example  # 作业的名字
  namespace: flink        # 指定在flink命名空间下运行
spec:
  image: flink:1.13.6      # Flink的镜像，改为使用1.13.6
  # 如果官方Flink镜像下载不了，可以使用此镜像
  # image: registry.cn-hangzhou.aliyuncs.com/cm_ns01/flink:1.13.6  
  flinkVersion: v1_13    # Flink的版本，与镜像版本保持一致
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
  serviceAccount: flink
  jobManager:
    resource:
      memory: "1024m"
      cpu: 1
  taskManager:
    resource:
      memory: "1024m"
      cpu: 1
  job:
    jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar  # Flink作业的启动类所在的Jar包路径
    parallelism: 2
    upgradeMode: stateless

在basic.yaml示例中，定义了一个Application模式的Flink作业（也可以称作Flink集群）规格信息，其中使用了1.13.6版本的Flink镜像。就如前面所述，Flink Kubernetes Operator支持的Flink最低版本就是1.13。此外，也定义了JobManager和TaskManager的资源配额，最后，在job部分，定义了Flink作业的启动类所在的Jar包路径，由于该Jar包默认就包含在Flink镜像里，所以可以直接使用本地路径方式读取。

2、运行示例Flink作业

kubectl apply -f basic.yaml

kubectl get all -n flink
kubectl get FlinkDeployment -n flink

可以看到，basic.yaml文件提交到K8s后，K8s在flink命名空间下新启动了2个Pod，一个是JobManager的Pod，名字是basic-example-556fd8bf6-tms8n，另一个是TaskManager的Pod，名字是basic-example-taskmanager-1-1。此外也可以看到创建了一个名字为basic-example，资源类型为FlinkDeployment的实例。

可以看到，basic.yaml文件提交到K8s后，K8s在flink命名空间下新启动了2个Pod，一个是JobManager的Pod，名字是basic-example-556fd8bf6-tms8n，另一个是TaskManager的Pod，名字是basic-example-taskmanager-1-1。此外也可以看到创建了一个名字为basic-example，资源类型为FlinkDeployment的实例。

3、终止示例Flink作业

停止Flink作业运行很简单，只需要执行以下任一命令即可。

kubectl delete -f basic.yam
kubectl delete FlinkDeployment basic-example -n flink

在下一篇文章中，会继续讲解和演示Flink Kubernetes Operator Application模式的2种Flink作业提交方式。更多有关Flink Kubernetes Operator的视频和学习资料，可以到bigdataonk8s观看和获取。