本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
# 使用 Amazon EMR on EKS 运行 Flink 任务
Amazon EMR 版本 6.13.0 及更高版本支持带 Apache Flink 的 Amazon EMR on EKS 或 Flink Kubernetes 运算符作为 Amazon EMR on EKS 的任务提交模型。借助带 Apache Flink 的 Amazon EMR on EKS，您可以在自己的 Amazon EKS 集群上使用 Amazon EMR 发行版运行时系统来部署和管理 Flink 应用程序。在 Amazon EKS 集群中部署 Flink Kubernetes 运算符后，您可以直接向此运算符提交 Flink 应用程序。运算符管理 Flink 应用程序的生命周期。
**Topics**
+ [设置和使用 Flink Kubernetes Operator](jobruns-flink-kubernetes-operator.md)
+ [使用 Flink Native Kubernetes](jobruns-flink-native-kubernetes.md)
+ [为 Flink 和 FluentD 自定义 Docker 映像](jobruns-flink-docker-flink-fluentd.md)
+ [监视 Flink Kubernetes Operator 和 Flink 任务](jobruns-flink-monitoring.md)
+ [Flink 如何支持高可用性和作业弹性](jobruns-flink-resiliency.md)
+ [在 Flink 应用程序中使用 Autoscaler](jobruns-flink-autoscaler.md)
+ [Amazon EMR on EKS 上 Flink 作业的维护和故障排除](jobruns-flink-troubleshooting.md)
+ [支持将 Amazon EMR on EKS 与 Apache Flink 结合使用的发行版](jobruns-flink-security-release-versions.md)
# 设置和使用 Flink Kubernetes Operator
以下页面旨在介绍如何在 Amazon EMR on EKS 上设置并使用 Flink Kubernetes Operator 来运行 Flink 任务。提供的主题包括所需的先决条件、如何设置环境以及在 Amazon EMR on EKS 上运行 Flink 应用程序。
**Topics**
+ [设置 Amazon EMR on EKS 的 Flink Kubernetes Operator](jobruns-flink-kubernetes-operator-setup.md)
+ [安装 Amazon EMR on EKS 的 Flink Kubernetes Operator](jobruns-flink-kubernetes-operator-getting-started.md)
+ [运行 Flink 应用程序](jobruns-flink-kubernetes-operator-run-application.md)
+ [运行 Flink 应用程序的安全角色权限](jobruns-flink-kubernetes-security.md)
+ [卸载 Amazon EMR on EKS 的 Flink Kubernetes Operator](jobruns-flink-kubernetes-operator-uninstall.md)
# 设置 Amazon EMR on EKS 的 Flink Kubernetes Operator
在 Amazon EKS 上安装 Flink Kubernetes Operator 之前，请执行下述任务来完成设置。如果已注册 Amazon Web Services（AWS）并且一直在使用 Amazon EKS，您基本上就准备好使用 Amazon EMR on EKS 了。完成以下任务，在 Amazon EKS 上设置好 Flink Operator。跳过已完成的先决条件，转到下一个先决条件。
+ **[安装或更新到最新版本的 AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html)** — 如果您已经安装了 AWS CLI，请确认您安装的是最新版本。
+ **[设置 kubectl 和 eksctl](https://docs.aws.amazon.com/eks/latest/userguide/install-kubectl.html)**：eksctl 是用来与 Amazon EKS 通信的命令行工具。
+ **[Install Helm](https://docs.aws.amazon.com/eks/latest/userguide/helm.html)**：Kubernetes 的 Helm 包管理器可帮助您在 Kubernetes 集群上安装和管理应用程序。
+ **[开始使用 Amazon EKS - eksctl](https://docs.aws.amazon.com/eks/latest/userguide/getting-started-eksctl.html)**：按照以下步骤在 Amazon EKS 中创建带有节点的新 Kubernetes 集群。
+ **[选择 Amazon EMR 发行版标签](jobruns-flink-security-release-versions.md)（6.13.0 或更高版本）**：Amazon EMR 6.13.0 及更高版本支持 Flink Kubernetes Operator。
+ **[在 Amazon EKS 集群上启用服务账户的 IAM 角色（IRSA）。](setting-up-enable-IAM.md)**
+ **[创建任务执行角色](creating-job-execution-role.md)**。
+ **[更新任务执行角色的信任策略。](setting-up-trust-policy.md)**
+ 创建 Operator 执行角色。此为可选步骤。Flink 任务和 Operator 可以使用相同的角色。如果想为 Operator 设置不同的 IAM 角色，可以单独创建一个角色。
+ 更新 Operator 执行角色的信任策略。必须为要用于 Amazon EMR Flink Kubernetes Operator 服务账户的角色显式添加一个信任策略条目。可以按照如下示例格式进行操作：
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"sts:AssumeRoleWithWebIdentity"
],
"Resource": [
"*"
],
"Condition": {
"StringLike": {
"aws:userid": "system:serviceaccount:emr:emr-containers-sa-flink-operator"
}
},
"Sid": "AllowSTSAssumerolewithwebidentity"
}
]
}
```
------
# 安装 Amazon EMR on EKS 的 Flink Kubernetes Operator
本主题通过准备 Flink 部署，帮助您开始在 Amazon EKS 上使用 Flink Kubernetes Operator。
## 安装 Kubernetes Operator
按照以下步骤安装 Apache Flink 版 Kubernetes Operator。
1. 如果尚未执行此操作，请完成 [设置 Amazon EMR on EKS 的 Flink Kubernetes Operator](jobruns-flink-kubernetes-operator-setup.md) 中的步骤。
1. 安装*cert-manager*（每个 Amazon EKS 集群安装一次）以启用添加 webhook 组件。
```
kubectl apply -f https://github.com/cert-manager/cert-manager/releases/download/v1.12.0/cert-manager.yaml
```
1. 安装 Helm 图表。
```
export VERSION=7.12.0 # The Amazon EMR release version
export NAMESPACE=The Kubernetes namespace to deploy the operator
helm install flink-kubernetes-operator \
oci://public.ecr.aws/emr-on-eks/flink-kubernetes-operator \
--version $VERSION \
--namespace $NAMESPACE
```
输出示例：
```
NAME: flink-kubernetes-operator
LAST DEPLOYED: Tue May 31 17:38:56 2022
NAMESPACE: $NAMESPACE
STATUS: deployed
REVISION: 1
TEST SUITE: None
```
1. 等待部署完成并验证图表安装情况。
```
kubectl wait deployment flink-kubernetes-operator --namespace $NAMESPACE --for condition=Available=True --timeout=30s
```
1. 部署完成后，您应该会看到如下消息。
```
deployment.apps/flink-kubernetes-operator condition met
```
1. 使用以下命令查看部署的 Operator。
```
helm list --namespace $NAMESPACE
```
示例输出如下，其中应用程序版本 `x.y.z-amzn-n` 将与您的 Amazon EMR on EKS 发行版的 Flink Operator 版本对应。有关更多信息，请参阅 [支持将 Amazon EMR on EKS 与 Apache Flink 结合使用的发行版](jobruns-flink-security-release-versions.md)。
```
NAME NAMESPACE REVISION UPDATED STATUS CHART APP VERSION
flink-kubernetes-operator $NAMESPACE 1 2023-02-22 16:43:45.24148 -0500 EST deployed flink-kubernetes-operator-emr-7.12.0 x.y.z-amzn-n
```
### 升级 Kubernetes Operator
要升级 Kubernetes Operator 的版本，请按照以下步骤进行操作：
1. 卸载旧的 `flink-kubernetes-operator`：`helm uninstall flink-kubernetes-operator -n `。
1. 删除 CRD（因为 Helm 不会自动删除旧的 CRD）：`kubectl delete crd flinkdeployments.flink.apache.org flinksessionjobs.flink.apache.org`。
1. 使用较新的版本重新安装 `flink-kubernetes-operator`。
# 运行 Flink 应用程序
在 Amazon EMR 6.13.0 及更高版本中，您可以在应用程序模式下在 EKS 上的 Amazon EMR 上使用 Flink Kubernetes Operator 来运行 Flink 应用程序。在 Amazon EMR 6.15.0 及更高版本中，您还可以在会话模式中运行 Flink 应用程序。本页介绍您使用 EKS 上的 Amazon EMR 运行 Flink 应用程序时可用的两种方法。
**Topics**
**注意**
提交 Flink 作业时，必须使用一个 Amazon S3 存储桶来存储高可用性元数据。如果不想使用此功能，可以将其禁用。系统会默认启用该功能。
**先决条件**：在使用 Flink Native Kubernetes Operator 运行 Flink 应用程序之前，请先完成 [设置 Amazon EMR on EKS 的 Flink Kubernetes Operator](jobruns-flink-kubernetes-operator-setup.md) 和 [安装 Kubernetes Operator](jobruns-flink-kubernetes-operator-getting-started.md#jobruns-flink-kubernetes-operator-getting-started-install-operator) 中的步骤。
------
#### [ Application mode ]
在 Amazon EMR 6.13.0 及更高版本中，您可以在应用程序模式下在 EKS 上的 Amazon EMR 上使用 Flink Kubernetes Operator 来运行 Flink 应用程序。
1. 创建一个名为 `basic-example-app-cluster.yaml` 的 `FlinkDeployment` 定义文件，如下例所示。如果您已激活并使用其中一个[选项 AWS 区域，请](https://docs.aws.amazon.com/controltower/latest/userguide/opt-in-region-considerations.html)务必取消注释并配置配置。`fs.s3a.endpoint.region`
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: basic-example-app-cluster
spec:
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
#fs.s3a.endpoint.region: OPT_IN_AWS_REGION_NAME
state.checkpoints.dir: CHECKPOINT_S3_STORAGE_PATH
state.savepoints.dir: SAVEPOINT_S3_STORAGE_PATH
flinkVersion: v1_17
executionRoleArn: JOB_EXECUTION_ROLE_ARN
emrReleaseLabel: "emr-6.13.0-flink-latest" # 6.13 or higher
jobManager:
storageDir: HIGH_AVAILABILITY_STORAGE_PATH
resource:
memory: "2048m"
cpu: 1
taskManager:
resource:
memory: "2048m"
cpu: 1
job:
# if you have your job jar in S3 bucket you can use that path as well
jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
parallelism: 2
upgradeMode: savepoint
savepointTriggerNonce: 0
monitoringConfiguration:
cloudWatchMonitoringConfiguration:
logGroupName: LOG_GROUP_NAME
```
1. 使用如下命令提交 Flink 部署。此操作还会创建一个名为 `basic-example-app-cluster` 的 `FlinkDeployment` 对象。
```
kubectl create -f basic-example-app-cluster.yaml -n
```
1. 访问 Flink UI。
```
kubectl port-forward deployments/basic-example-app-cluster 8081 -n NAMESPACE
```
1. 打开 `localhost:8081` 即可在本地查看 Flink 任务。
1. 清理任务。记得清理为此任务创建的 S3 项目，例如检查点、高可用性、保存点元数据和日志。 CloudWatch
有关通过 Flink Kubernetes 运算符向 Flink 提交应用程序的更多信息，请参阅文件夹中的 Flink K [uber](https://github.com/apache/flink-kubernetes-operator/tree/main/examples) netes 运算符示例。`apache/flink-kubernetes-operator` GitHub
------
#### [ Session mode ]
在 Amazon EMR 6.15.0 及更高版本中，您可以在会话模式下在 EKS 上的 Amazon EMR 上使用 Flink Kubernetes Operator 来运行 Flink 应用程序。
1. 创建一个名为 `basic-example-app-cluster.yaml` 的 `FlinkDeployment` 定义文件，如下例所示。如果您已激活并使用其中一个[选项 AWS 区域，请](https://docs.aws.amazon.com/controltower/latest/userguide/opt-in-region-considerations.html)务必取消注释并配置配置。`fs.s3a.endpoint.region`
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: basic-example-session-cluster
spec:
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
#fs.s3a.endpoint.region: OPT_IN_AWS_REGION_NAME
state.checkpoints.dir: CHECKPOINT_S3_STORAGE_PATH
state.savepoints.dir: SAVEPOINT_S3_STORAGE_PATH
flinkVersion: v1_17
executionRoleArn: JOB_EXECUTION_ROLE_ARN
emrReleaseLabel: "emr-6.15.0-flink-latest"
jobManager:
storageDir: HIGH_AVAILABILITY_S3_STORAGE_PATH
resource:
memory: "2048m"
cpu: 1
taskManager:
resource:
memory: "2048m"
cpu: 1
monitoringConfiguration:
s3MonitoringConfiguration:
logUri:
cloudWatchMonitoringConfiguration:
logGroupName: LOG_GROUP_NAME
```
1. 使用如下命令提交 Flink 部署。此操作还会创建一个名为 `basic-example-session-cluster` 的 `FlinkDeployment` 对象。
```
kubectl create -f basic-example-app-cluster.yaml -n NAMESPACE
```
1. 使用以下命令确认会话集群 `LIFECYCLE` 是 `STABLE`：
```
kubectl get flinkdeployments.flink.apache.org basic-example-session-cluster -n NAMESPACE
```
该输出应该类似于以下示例：
```
NAME JOB STATUS LIFECYCLE STATE
basic-example-session-cluster STABLE
```
1. 使用以下示例内容创建 `FlinkSessionJob` 自定义定义资源文件 `basic-session-job.yaml`：
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkSessionJob
metadata:
name: basic-session-job
spec:
deploymentName: basic-session-deployment
job:
# If you have your job jar in an S3 bucket you can use that path.
# To use jar in S3 bucket, set
# OPERATOR_EXECUTION_ROLE_ARN (--set emrContainers.operatorExecutionRoleArn=$OPERATOR_EXECUTION_ROLE_ARN)
# when you install Spark operator
jarURI: https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming_2.12/1.16.1/flink-examples-streaming_2.12-1.16.1-TopSpeedWindowing.jar
parallelism: 2
upgradeMode: stateless
```
1. 使用如下命令提交 Flink 会话作业。此操作将会创建一个 `FlinkSessionJob` 对象 `basic-session-job`。
```
kubectl apply -f basic-session-job.yaml -n $NAMESPACE
```
1. 使用以下命令确认会话集群 `LIFECYCLE` 是 `STABLE`，`JOB STATUS` 是 `RUNNING`：
```
kubectl get flinkdeployments.flink.apache.org basic-example-session-cluster -n NAMESPACE
```
该输出应该类似于以下示例：
```
NAME JOB STATUS LIFECYCLE STATE
basic-example-session-cluster RUNNING STABLE
```
1. 访问 Flink UI。
```
kubectl port-forward deployments/basic-example-session-cluster 8081 -n NAMESPACE
```
1. 打开 `localhost:8081` 即可在本地查看 Flink 任务。
1. 清理任务。记得清理为此任务创建的 S3 项目，例如检查点、高可用性、保存点元数据和日志。 CloudWatch
------
# 运行 Flink 应用程序的安全角色权限
本主题介绍了部署和运行 Flink 应用程序的安全角色。管理部署以及创建和管理作业需要两个角色：Operator 和 Job。本主题介绍了两个角色并列出了其权限。
## 基于角色的访问控制
要部署 Operator 并运行 Flink 任务，必须创建两个 Kubernetes 角色：一个 Operator 角色和一个任务角色。安装 Operator 时 Amazon EMR 会默认创建这两个角色。
## Operator 角色
我们使用操作员角色`flinkdeployments`来管理每个 Flink 作业和其他资源（例如服务）的创建和管理。 JobManager
Operator 角色的默认名称为 `emr-containers-sa-flink-operator` 并且需要以下权限。
```
rules:
- apiGroups:
- ""
resources:
- pods
- services
- events
- configmaps
- secrets
- serviceaccounts
verbs:
- '*'
- apiGroups:
- rbac.authorization.k8s.io
resources:
- roles
- rolebindings
verbs:
- '*'
- apiGroups:
- apps
resources:
- deployments
- deployments/finalizers
- replicasets
verbs:
- '*'
- apiGroups:
- extensions
resources:
- deployments
- ingresses
verbs:
- '*'
- apiGroups:
- flink.apache.org
resources:
- flinkdeployments
- flinkdeployments/status
- flinksessionjobs
- flinksessionjobs/status
verbs:
- '*'
- apiGroups:
- networking.k8s.io
resources:
- ingresses
verbs:
- '*'
- apiGroups:
- coordination.k8s.io
resources:
- leases
verbs:
- '*'
```
## 任务角色
JobManager 使用作业角色 ConfigMaps 为每个作业创建和管理 TaskManagers 和。
```
rules:
- apiGroups:
- ""
resources:
- pods
- configmaps
verbs:
- '*'
- apiGroups:
- apps
resources:
- deployments
- deployments/finalizers
verbs:
- '*'
```
# 卸载 Amazon EMR on EKS 的 Flink Kubernetes Operator
按照如下步骤卸载 Flink Kubernetes Operator。
1. 删除 Operator。
```
helm uninstall flink-kubernetes-operator -n
```
1. 删除 Helm 未卸载的 Kubernetes 资源。
```
kubectl delete serviceaccounts, roles, rolebindings -l emr-containers.amazonaws.com/component=flink.operator --namespace
kubectl delete crd flinkdeployments.flink.apache.org flinksessionjobs.flink.apache.org
```
1. （可选）删除 cert-manager。
```
kubectl delete -f https://github.com/jetstack/cert-manager/releases/download/v1.12.0/cert-manager.yaml
```
# 使用 Flink Native Kubernetes
Amazon EMR 6.13.0 及更高版本都支持 Flink Native Kubernetes 作为命令行工具，而您可以使用该工具向 Amazon EMR on EKS 集群提交和执行 Flink 应用程序。
**Topics**
+ [设置 Amazon EMR on EKS 的 Flink Native Kubernetes](jobruns-flink-native-kubernetes-setup.md)
+ [Amazon EMR on EKS 的 Flink Native Kubernetes 入门](jobruns-flink-native-kubernetes-getting-started.md)
+ [原生 Kubernetes 的 Flink JobManager 服务账号安全要求](jobruns-flink-native-kubernetes-security-requirements.md)
# 设置 Amazon EMR on EKS 的 Flink Native Kubernetes
先执行以下任务来完成设置，才能在 Amazon EMR on EKS 上使用 Flink CLI 运行应用程序。如果已注册 Amazon Web Services（AWS）并且一直在使用 Amazon EKS，您基本上就准备好使用 Amazon EMR on EKS 了。跳过已完成的先决条件，转到下一个先决条件。
+ **[安装或更新到最新版本的 AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html)** — 如果您已经安装了 AWS CLI，请确认您安装的是最新版本。
+ **[开始使用 Amazon EKS - eksctl](https://docs.aws.amazon.com/eks/latest/userguide/getting-started-eksctl.html)**：按照以下步骤在 Amazon EKS 中创建带有节点的新 Kubernetes 集群。
+ **[选择 Amazon EMR 基础映像 URI](docker-custom-images-tag.md)（6.13.0 或更高版本）**：Amazon EMR 6.13.0 及更高版本都支持 Flink Kubernetes 命令。
+ 确认 JobManager 服务帐号具有创建和监视 TaskManager pod 的相应权限。有关更多信息，请参阅[原生 Kubernetes 的 Flink JobManager 服务账号安全要求](https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/jobruns-flink-native-kubernetes-security-requirements.html)。
+ 设置本地 [AWS 凭证配置文件](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-files.html)。
+ [为 Amazon EKS 集群创建或更新 kubeconfig 文件](https://docs.aws.amazon.com/eks/latest/userguide/create-kubeconfig.html)，如果要在该集群上运行 Flink 应用程序。
# Amazon EMR on EKS 的 Flink Native Kubernetes 入门
这些步骤展示了如何配置和运行 Flink 应用程序并为其设置服务账户。Flink Native Kubernetes 用于在运行的 Kubernetes 集群上部署 Flink。
## 配置和运行 Flink 应用程序
Amazon EMR 6.13.0 及更高版本都支持 Flink Native Kubernetes 在 Amazon EKS 集群上运行 Flink 应用程序。要运行 Flink 应用程序，请按照下述步骤操作：
1. 在使用 Flink Native Kubernetes 命令运行 Flink 应用程序之前，请先完成 [设置 Amazon EMR on EKS 的 Flink Native Kubernetes](jobruns-flink-native-kubernetes-setup.md) 中的步骤。
1. [下载并安装 Flink](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/try-flink/local_installation)。
1. 设置以下环境变量的值。
```
#Export the FLINK_HOME environment variable to your local installation of Flink
export FLINK_HOME=/usr/local/bin/flink #Will vary depending on your installation
export NAMESPACE=flink
export CLUSTER_ID=flink-application-cluster
export IMAGE=<123456789012.dkr.ecr.sample-AWS 区域-.amazonaws.com/flink/emr-6.13.0-flink:latest>
export FLINK_SERVICE_ACCOUNT=emr-containers-sa-flink
export FLINK_CLUSTER_ROLE_BINDING=emr-containers-crb-flink
```
1. 创建服务账户来管理 Kubernetes 资源。
```
kubectl create serviceaccount $FLINK_SERVICE_ACCOUNT -n $NAMESPACE
kubectl create clusterrolebinding $FLINK_CLUSTER_ROLE_BINDING --clusterrole=edit --serviceaccount=$NAMESPACE:$FLINK_SERVICE_ACCOUNT
```
1. 运行 `run-application` CLI 命令。
```
$FLINK_HOME/bin/flink run-application \
--target kubernetes-application \
-Dkubernetes.namespace=$NAMESPACE \
-Dkubernetes.cluster-id=$CLUSTER_ID \
-Dkubernetes.container.image.ref=$IMAGE \
-Dkubernetes.service-account=$FLINK_SERVICE_ACCOUNT \
local:///opt/flink/examples/streaming/Iteration.jar
2022-12-29 21:13:06,947 INFO org.apache.flink.kubernetes.utils.KubernetesUtils [] - Kubernetes deployment requires a fixed port. Configuration blob.server.port will be set to 6124
2022-12-29 21:13:06,948 INFO org.apache.flink.kubernetes.utils.KubernetesUtils [] - Kubernetes deployment requires a fixed port. Configuration taskmanager.rpc.port will be set to 6122
2022-12-29 21:13:07,861 WARN org.apache.flink.kubernetes.KubernetesClusterDescriptor [] - Please note that Flink client operations(e.g. cancel, list, stop, savepoint, etc.) won't work from outside the Kubernetes cluster since 'kubernetes.rest-service.exposed.type' has been set to ClusterIP.
2022-12-29 21:13:07,868 INFO org.apache.flink.kubernetes.KubernetesClusterDescriptor [] - Create flink application cluster flink-application-cluster successfully, JobManager Web Interface: http://flink-application-cluster-rest.flink:8081
```
1. 检查创建好的 Kubernetes 资源。
```
kubectl get all -n
NAME READY STATUS RESTARTS AGE
pod/flink-application-cluster-546687cb47-w2p2z 1/1 Running 0 3m37s
pod/flink-application-cluster-taskmanager-1-1 1/1 Running 0 3m24s
NAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGE
service/flink-application-cluster ClusterIP None 6123/TCP,6124/TCP 3m38s
service/flink-application-cluster-rest ClusterIP 10.100.132.158 8081/TCP 3m38s
NAME READY UP-TO-DATE AVAILABLE AGE
deployment.apps/flink-application-cluster 1/1 1 1 3m38s
NAME DESIRED CURRENT READY AGE
replicaset.apps/flink-application-cluster-546687cb47 1 1 1 3m38s
```
1. 端口转发到 8081。
```
kubectl port-forward service/flink-application-cluster-rest 8081 -n
Forwarding from 127.0.0.1:8081 -> 8081
```
1. 在本地访问 Flink UI。
![\[访问 Flink UI。\]](http://docs.aws.amazon.com/zh_cn/emr/latest/EMR-on-EKS-DevelopmentGuide/images/jobruns-flink-native-kubernetes-ui.png)
1. 删除 Flink 应用程序。
```
kubectl delete deployment.apps/flink-application-cluster -n
deployment.apps "flink-application-cluster" deleted
```
有关向 Flink 提交应用程序的更多信息，请参阅 Apache Flink 文档中的 [Native Kubernetes](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/resource-providers/native_kubernetes/)。
# 原生 Kubernetes 的 Flink JobManager 服务账号安全要求
Flink JobManager 容器使用 Kubernetes 服务账号访问 Kubernetes API 服务器来创建和监视 Pod。 TaskManager JobManager 服务帐号必须拥有对 create/delete TaskManager pod 的相应权限，并允许 to watch 领导者 ConfigMaps 检索集群 ResourceManager 中 JobManager 和的地址。 TaskManager
以下规则适用于此服务账户。
```
rules:
- apiGroups:
- ""
resources:
- pods
verbs:
- "*"
- apiGroups:
- ""
resources:
- services
verbs:
- "*"
- apiGroups:
- ""
resources:
- configmaps
verbs:
- "*"
- apiGroups:
- "apps"
resources:
- deployments
verbs:
- "*"
```
# 为 Flink 和 FluentD 自定义 Docker 映像
按照以下步骤，使用 Apache Flink 或 FluentD 映像为 Amazon EMR on EKS 自定义 Docker 映像。其中包括获取基础映像、自定义基础映像、发布基础映像和提交工作负载的技术指导。
**Topics**
+ [先决条件](#jobruns-flink-docker-flink-fluentd-prereqs)
+ [步骤 1：从 Amazon Elastic Container Registry 中检索基础映像](#jobruns-flink-docker-flink-fluentd-retrieve-base)
+ [步骤 2：自定义基础镜像](#jobruns-flink-docker-flink-fluentd-customize-image)
+ [步骤 3：发布自定义映像](#jobruns-flink-docker-flink-fluentd-publish-image)
+ [步骤 4：使用自定义映像在 Amazon EMR 中提交 Flink 工作负载](#jobruns-flink-docker-flink-fluentd-submit-workload)
## 先决条件
在自定义 Docker 映像之前，确保您已完成以下先决条件：
+ 已完成[设置 Amazon EMR on EKS 的 Flink Kubernetes Operator](https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/jobruns-flink-kubernetes-operator-setup.html) 步骤。
+ 已在环境中安装 Docker。有关更多信息，请参阅[获取 Docker](https://docs.docker.com/get-docker/)。
## 步骤 1：从 Amazon Elastic Container Registry 中检索基础映像
基础映像包含访问其他 AWS 服务所需的 Amazon EMR 运行时和连接器。如果在 Flink 6.14.0 或更高版本中使用 Amazon EMR on EKS，可从 Amazon ECR 公开映像浏览馆获取基础映像。浏览图库以找到映像链接，然后将映像拉到本地 Workspace。例如，对于 Amazon EMR 6.14.0 发行版，以下 `docker pull` 命令将返回最新的标准基础映像。将 `emr-6.14.0:latest` 替换为所需的发行版。
```
docker pull public.ecr.aws/emr-on-eks/flink/emr-6.14.0-flink:latest
```
以下是 Flink 库映像和 Fluentd 库映像的链接：
+ [emr-on-eks/flink/emr-6.14.0-flink](https://gallery.ecr.aws/emr-on-eks/flink/emr-6.14.0-flink)
+ [emr-on-eks/fluentd/emr-6.14.0 (](https://gallery.ecr.aws/emr-on-eks/fluentd/emr-6.14.0)
## 步骤 2：自定义基础镜像
下列步骤描述了如何自定义从 Amazon ECR 中拉取的基础映像。
1. 在您的本地 Workspace 上创建新的 `Dockerfile`。
1. 编辑 `Dockerfile` 并添加以下内容。该 `Dockerfile` 使用您从 `public.ecr.aws/emr-on-eks/flink/emr-7.12.0-flink:latest` 中拉取的容器映像。
```
FROM public.ecr.aws/emr-on-eks/flink/emr-7.12.0-flink:latest
USER root
### Add customization commands here ####
USER hadoop:hadoop
```
如果您使用的是 `Fluentd`，请使用以下配置。
```
FROM public.ecr.aws/emr-on-eks/fluentd/emr-7.12.0:latest
USER root
### Add customization commands here ####
USER hadoop:hadoop
```
1. 将命令添加到 `Dockerfile` 以自定义基础镜像。以下命令演示了如何安装 Python 库。
```
FROM public.ecr.aws/emr-on-eks/flink/emr-7.12.0-flink:latest
USER root
RUN pip3 install --upgrade boto3 pandas numpy // For python 3
USER hadoop:hadoop
```
1. 在创建 `DockerFile` 的同一目录中，运行以下命令构建 Docker 映像。您在 `-t` 标志后提供的字段是您自定义的映像名称。
```
docker build -t .dkr.ecr..amazonaws.com/:
```
## 步骤 3：发布自定义映像
现在，将新的 Docker 映像发布到您的 Amazon ECR 注册表。
1. 运行以下命令创建一个 Amazon ECR 存储库，来存储您的 Docker 映像。为存储库命名，例如 `emr_custom_repo.`。有关更多信息，请参阅《Amazon Elastic Container Registry 用户指南》中的[创建存储库](https://docs.aws.amazon.com/AmazonECR/latest/userguide/getting-started-cli.html#cli-create-repository)。
```
aws ecr create-repository \
--repository-name emr_custom_repo \
--image-scanning-configuration scanOnPush=true \
--region
```
1. 运行以下命令对您的默认注册表进行身份验证。有关更多信息，请参阅《Amazon Elastic Container Registry 用户指南》中的[对您的默认注册表进行身份验证](https://docs.aws.amazon.com/AmazonECR/latest/userguide/getting-started-cli.html#cli-authenticate-registry)。
```
aws ecr get-login-password --region | docker login --username AWS --password-stdin .dkr.ecr..amazonaws.com
```
1. 推送镜像。有关更多信息，请参阅《Amazon Elastic Container Registry 用户指南》中的[将映像推送到 Amazon ECR](https://docs.aws.amazon.com/AmazonECR/latest/userguide/getting-started-cli.html#cli-push-image)。
```
docker push .dkr.ecr..amazonaws.com/:
```
## 步骤 4：使用自定义映像在 Amazon EMR 中提交 Flink 工作负载
要使用自定映像，请对 `FlinkDeployment` 规范进行以下更改。为此，请在部署规范的 `spec.image` 行中输入您的映像。
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: basic-example
spec:
flinkVersion: v1_18
image: .dkr.ecr..amazonaws.com/:
imagePullPolicy: Always
flinkConfiguration:
taskmanager.numberOfTaskSlots: "1"
```
要在 Fluentd 作业中使用自定义映像，请在部署规范的 `monitoringConfiguration.image` 行中输入您的映像。
```
monitoringConfiguration:
image: .dkr.ecr..amazonaws.com/:
cloudWatchMonitoringConfiguration:
logGroupName: flink-log-group
logStreamNamePrefix: custom-fluentd
```
# 监视 Flink Kubernetes Operator 和 Flink 任务
本节旨在介绍在 Amazon EMR on EKS 上监控 Flink 任务的几种方法。其中包括将 Flink 与 Amazon Managed Service for Prometheus 集成，使用提供作业状态和指标的 *Flink Web 控制面板*，或者使用监控配置将日志数据发送到 Amazon S3 和 Amazon CloudWatch。
**Topics**
+ [使用 Amazon Managed Service for Prometheus 监控 Flink 作业](jobruns-flink-monitoring-prometheus.md)
+ [使用 Flink UI 监控 Flink 作业](jobruns-flink-monitoring-ui.md)
+ [使用监控配置监控 Flink Kubernetes Operator 和 Flink 作业](jobruns-flink-monitoring-configuration.md)
# 使用 Amazon Managed Service for Prometheus 监控 Flink 作业
您可以将 Apache Flink 与 Amazon Managed Service for Prometheus（管理门户）集成。Amazon Managed Service for Prometheus 支持从在 Amazon EKS 上运行的集群中摄取 Amazon Managed Service for Prometheus 服务器的指标。Amazon Managed Service for Prometheus 可与已 Amazon EKS 集群上运行的 Prometheus 服务器配合使用。运行集成了 Amazon EMR Flink Operator 的 Amazon Managed Service for Prometheus 会自动部署并配置 Prometheus 服务器，使其与 Amazon Managed Service for Prometheus 集成。
1. [创建 Amazon Managed Service for Prometheus Workspace](https://docs.aws.amazon.com/prometheus/latest/userguide/AMP-onboard-create-workspace.html)。此工作空间用作摄取端点。稍后需要远程写入 URL。
1. 设置服务账户的 IAM 角色。
要使用这种入门方法，请使用运行 Prometheus 服务器的 Amazon EKS 集群中服务账户的 IAM 角色。此类角色又称为*服务角色*。
如果尚无此类角色，请[设置服务角色从 Amazon EKS 集群中摄取指标](https://docs.aws.amazon.com/prometheus/latest/userguide/set-up-irsa.html)。
请创建一个名为 `amp-iamproxy-ingest-role` 的 IAM 角色，然后再继续操作。
1. 使用 Amazon Managed Service for Prometheus 安装 Amazon EMR Flink Operator。
您已经拥有了一个 Amazon Managed Service for Prometheus Workspace，一个专用于 Amazon Managed Service for Prometheus 的 IAM 角色，以及所需的权限，现在可以安装 Amazon EMR Flink 运算符。
创建 `enable-amp.yaml` 文件。此文件允许您使用自定义配置来覆盖 Amazon Managed Service for Prometheus 的设置。务必使用您自己的角色。
```
kube-prometheus-stack:
prometheus:
serviceAccount:
create: true
name: "amp-iamproxy-ingest-service-account"
annotations:
eks.amazonaws.com/role-arn: "arn:aws:iam:::role/amp-iamproxy-ingest-role"
remoteWrite:
- url:
sigv4:
region:
queueConfig:
maxSamplesPerSend: 1000
maxShards: 200
capacity: 2500
```
使用 [https://helm.sh/docs/helm/helm_install/](https://helm.sh/docs/helm/helm_install/) 命令将覆盖传递给 `flink-kubernetes-operator` 图表。
```
helm upgrade -n flink-kubernetes-operator \
oci://public.ecr.aws/emr-on-eks/flink-kubernetes-operator \
--set prometheus.enabled=true
-f enable-amp.yaml
```
此命令会自动在端口 9999 的 Operator 上安装 Prometheus 报告程序。未来的任何 `FlinkDeployment` 也会在 9249 上公开一个 `metrics` 端口。
+ Flink Operator 指标会出现在 Prometheus 中的 `flink_k8soperator_` 标签下。
+ Flink Task Manager 指标会出现在 Prometheus 的 `flink_taskmanager_` 标签下。
+ Flink Job Manager 指标会出现在 Prometheus 的 `flink_jobmanager_` 标签下。
# 使用 Flink UI 监控 Flink 作业
要监控正在运行的 Flink 应用程序的运行状况和性能，请使用 *Flink Web 控制面板*。此仪表板提供有关任务状态 TaskManagers、数量以及该任务的指标和日志的信息。借助它还可以查看并修改 Flink 作业配置，以及通过提交或取消作业与 Flink 集群进行交互。
要访问 Kubernetes 上正在运行的 Flink 应用程序的 Flink Web 控制面板，请按照以下步骤操作：
1. 使用`kubectl port-forward`命令将本地端口转发到 Flink 应用程序容器中运行 Flink Web 控制面板的 TaskManager 端口。默认情况下，此端口为 8081。*deployment-name*替换为上面的 Flink 应用程序部署名称。
```
kubectl get deployments -n namespace
```
输出示例：
```
kubectl get deployments -n flink-namespace
NAME READY UP-TO-DATE AVAILABLE AGE
basic-example 1/1 1 1 11m
flink-kubernetes-operator 1/1 1 1 21h
```
```
kubectl port-forward deployments/deployment-name 8081 -n namespace
```
1. 如果要在本地使用其他端口，请使用:8081 *local-port* 参数。
```
kubectl port-forward -n flink deployments/basic-example 8080:8081
```
1. 在 Web 浏览器中导航到 `http://localhost:8081`（如果使用的是自定义本地端口，则导航到 `http://localhost:local-port`）来访问 Flink Web 控制面板。此仪表板显示有关正在运行的 Flink 应用程序的信息，例如作业的状态 TaskManagers、数量以及该作业的指标和日志。
![\[Flink 控制面板用户界面示例\]](http://docs.aws.amazon.com/zh_cn/emr/latest/EMR-on-EKS-DevelopmentGuide/images/sample-flink-dashboard-ui.png)
# 使用监控配置监控 Flink Kubernetes Operator 和 Flink 作业
监控配置使您可以轻松地将 Flink 应用程序和操作员日志的日志存档设置为 S3 and/or CloudWatch （您可以选择其中一个或两个）。这样做会向 JobManager 你的 TaskManager 和容器添加一个 FluentD 边车，然后将这些组件的日志转发到你配置的接收器。
**注意**
必须为 Flink Operator 和 Flink 任务（服务账户）设置服务账户的 IAM 角色，才能使用此功能，因为它需要与其他 AWS 服务进行交互。您必须按照 [设置 Amazon EMR on EKS 的 Flink Kubernetes Operator](jobruns-flink-kubernetes-operator-setup.md)，使用 IRSA 进行设置。
## Flink 应用程序日志
您可以通过下列方式来定义此配置。
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: basic-example
spec:
image: FLINK IMAGE TAG
imagePullPolicy: Always
flinkVersion: v1_17
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
executionRoleArn: JOB EXECUTION ROLE
jobManager:
resource:
memory: "2048m"
cpu: 1
taskManager:
resource:
memory: "2048m"
cpu: 1
job:
jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
monitoringConfiguration:
s3MonitoringConfiguration:
logUri: S3 BUCKET
cloudWatchMonitoringConfiguration:
logGroupName: LOG GROUP NAME
logStreamNamePrefix: LOG GROUP STREAM PREFIX
sideCarResources:
limits:
cpuLimit: 500m
memoryLimit: 250Mi
containerLogRotationConfiguration:
rotationSize: 2GB
maxFilesToKeep: 10
```
配置选项如下。
+ `s3MonitoringConfiguration`：用来设置转发到 S3 的配置密钥
+ `logUri`（必需）：用来存储日志的 S3 存储桶路径。
+ 上传日志后，S3 中的路径如下所示。
+ 未启用日志轮换：
```
s3://${logUri}/${POD NAME}/STDOUT or STDERR.gz
```
+ 已启用日志轮换。您可以同时使用轮换文件和当前文件（不带日期戳的文件）。
```
s3://${logUri}/${POD NAME}/STDOUT or STDERR.gz
```
以下格式是递增数字。
```
s3://${logUri}/${POD NAME}/stdout_YYYYMMDD_index.gz
```
+ 使用此转发器需要以下 IAM 权限。
```
{
"Effect": "Allow",
"Action": [
"s3:PutObject"
],
"Resource": [
"S3_BUCKET_URI/*",
"S3_BUCKET_URI"
]
}
```
+ `cloudWatchMonitoringConfiguration`— 用于设置转发的配置密钥 CloudWatch。
+ `logGroupName`（必填）— 要向其发送 CloudWatch 日志的日志组的名称（如果该组不存在，则自动创建该组）。
+ `logStreamNamePrefix`（可选）– 要向其发送日志的日志流的名称。默认值是空字符串。格式如下所示：
```
${logStreamNamePrefix}/${POD NAME}/STDOUT or STDERR
```
+ 使用此转发器需要以下 IAM 权限。
```
{
"Effect": "Allow",
"Action": [
"logs:CreateLogStream",
"logs:CreateLogGroup",
"logs:PutLogEvents"
],
"Resource": [
"arn:aws:logs:REGION:ACCOUNT-ID:log-group:{YOUR_LOG_GROUP_NAME}:*",
"arn:aws:logs:REGION:ACCOUNT-ID:log-group:{YOUR_LOG_GROUP_NAME}"
]
}
```
+ `sideCarResources`（可选）：用于在启动的 Fluentbit Sidecar 容器上设置资源限制的配置密钥。
+ `memoryLimit`（可选）：默认值为 512Mi。根据自身需求进行调整。
+ `cpuLimit`（可选）：此选项没有默认值。根据自身需求进行调整。
+ `containerLogRotationConfiguration`（可选）：控制容器日志的轮换行为。该功能默认已启用。
+ `rotationSize`（必需）：指定日志轮换的文件大小。可行值的范围从 2KB 到 2GB 不等。rotationSize 参数的数字单位部分以整数形式传递。由于不支持十进制值，您可以指定 1.5GB 的轮换大小，例如值 1500MB。默认值为 2GB。
+ `maxFilesToKeep`（必需）：指定轮换后要在容器中保留的最大文件数。最小值为 1，最大值为 50。默认值为 10。
## Flink Operator 日志
我们还可以使用 Helm 图表安装 `values.yaml` 文件中的以下选项，为 Operator 启用日志存档。您可以启用 S3 CloudWatch、或两者兼而有之。
```
monitoringConfiguration:
s3MonitoringConfiguration:
logUri: "S3-BUCKET"
totalFileSize: "1G"
uploadTimeout: "1m"
cloudWatchMonitoringConfiguration:
logGroupName: "flink-log-group"
logStreamNamePrefix: "example-job-prefix-test-2"
sideCarResources:
limits:
cpuLimit: 1
memoryLimit: 800Mi
memoryBufferLimit: 700M
```
`monitoringConfiguration` 下的可用配置选项如下。
+ `s3MonitoringConfiguration`：设置此选项以存档到 S3。
+ `logUri`（必填项）：用来存储日志的 S3 存储桶路径。
+ 以下是上传日志后 S3 存储桶路径的示例格式。
+ 未启用日志轮换。
```
s3://${logUri}/${POD NAME}/OPERATOR or WEBHOOK/STDOUT or STDERR.gz
```
+ 已启用日志轮换。您可以同时使用轮换文件和当前文件（不带日期戳的文件）。
```
s3://${logUri}/${POD NAME}/OPERATOR or WEBHOOK/STDOUT or STDERR.gz
```
以下格式索引是递增数字。
```
s3://${logUri}/${POD NAME}/OPERATOR or WEBHOOK/stdout_YYYYMMDD_index.gz
```
+ `cloudWatchMonitoringConfiguration`— 用于设置转发的配置密钥 CloudWatch。
+ `logGroupName`（必填）— 您要向其发送 CloudWatch 日志的日志组的名称。如果日志组不存在，则会自动创建。
+ `logStreamNamePrefix`（可选）：要向其发送日志的日志流的名称。默认值是空字符串。中的格式 CloudWatch 如下：
```
${logStreamNamePrefix}/${POD NAME}/STDOUT or STDERR
```
+ `sideCarResources`（可选）：用于在启动的 Fluentbit Sidecar 容器上设置资源限制的配置密钥。
+ `memoryLimit`（可选）：内存限制。根据自身需求进行调整。默认值为 512Mi。
+ `cpuLimit`：CPU 限制。根据自身需求进行调整。无默认值。
+ `containerLogRotationConfiguration`（可选）：控制容器日志的轮换行为。该功能默认已启用。
+ `rotationSize`（必需）：指定日志轮换的文件大小。可行值的范围从 2KB 到 2GB 不等。rotationSize 参数的数字单位部分以整数形式传递。由于不支持十进制值，您可以指定 1.5GB 的轮换大小，例如值 1500MB。默认值为 2GB。
+ `maxFilesToKeep`（必需）：指定轮换后要在容器中保留的最大文件数。最小值为 1，最大值为 50。默认值为 10。
# Flink 如何支持高可用性和作业弹性
以下部分概述了 Flink 如何使作业更加可靠和高度可用。Flink 通过内置的高可用性功能和各种故障恢复功能来实现这一点。
**Topics**
+ [为 Flink Operator 和 Flink 应用程序使用高可用性（HA）](jobruns-flink-using-ha.md)
+ [使用 EKS 上的 Amazon EMR 优化 Flink 任务重启时间以进行任务恢复和扩展操作](jobruns-flink-restart.md)
+ [使用 EKS 上的 Amazon EMR 上的 Flink 正常停用竞价型实例](jobruns-flink-decommission.md)
# 为 Flink Operator 和 Flink 应用程序使用高可用性（HA）
本主题介绍了如何配置高可用性以及如何在几种不同的用例中使用。包括在使用 Job Manager 和 Flink Native Kubernetes 时。
## Flink Operator 高可用性
我们为 Flink Operator 启用了*高可用性*，这样就可以使用备用 Flink Operator 进行故障转移，从而在发生故障时最大限度地减少 Operator 控制回路中的停机时间。默认会启用“高可用性”，启动 Operator 副本的默认数量为 2。您可以在 Helm 图表的 `values.yaml` 文件中配置副本字段。
以下字段支持自定义：
+ `replicas`（可选，默认值为 2）：将此数字设置为大于 1 会创建其他备用 Operator，从而更快地恢复任务。
+ `highAvailabilityEnabled`（可选，默认值为 true）：控制是否要启用 HA。将此参数指定为 true 可启用多可用区部署支持，并设置正确的 `flink-conf.yaml` 参数。
在 `values.yaml` 文件中设置以下配置可以为 Operator 禁用 HA。
```
...
imagePullSecrets: []
replicas: 1
# set this to false if you don't want HA
highAvailabilityEnabled: false
...
```
**多可用区部署**
我们在多个可用区中创 Operator Pod。这是一个软约束，如果不同可用区中没有足够的资源，您的 Operator Pod 将被调度到同一可用区中。
**确定主副本**
如果启用了 HA，则副本使用租约来确定哪个 JMs 是领导者，并使用 K8s Lease 进行领导者选举。您可以描述租约并查看 .Spec.Holder Identity 字段来确定当前主副本
```
kubectl describe lease --lease -n | grep "Holder Identity"
```
**Flink-S3 交互**
**配置访问凭证**
请确保已为 IRSA 配置了相应的 IAM 权限来访问 S3 存储桶。
**从 S3 应用程序模式获取任务 jar**
Flink Operator 也支持从 S3 获取应用程序 jar。您只需在 FlinkDeployment 规范中提供 JarurI 的 S3 位置即可。
您也可以使用此功能下载其他工件，例如 PyFlink 脚本。生成的 Python 脚本放在路径 `/opt/flink/usrlib/` 下。
以下示例演示了如何将此功能用于作 PyFlink 业。注意 jarURI 和 args 字段。
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: python-example
spec:
image:
emrReleaseLabel: "emr-6.12.0-flink-latest"
flinkVersion: v1_16
flinkConfiguration:
taskmanager.numberOfTaskSlots: "1"
serviceAccount: flink
jobManager:
highAvailabilityEnabled: false
replicas: 1
resource:
memory: "2048m"
cpu: 1
taskManager:
resource:
memory: "2048m"
cpu: 1
job:
jarURI: "s3:///scripts/pyflink.py" # Note, this will trigger the artifact download process
entryClass: "org.apache.flink.client.python.PythonDriver"
args: ["-pyclientexec", "/usr/local/bin/python3", "-py", "/opt/flink/usrlib/pyflink.py"]
parallelism: 1
upgradeMode: stateless
```
**Flink S3 连接器**
Flink 随附两个 S3 连接器（如下所示）。以下各节旨在介绍何时使用哪个连接器。
**检查点：Presto S3 连接器**
+ 将 S3 方案设置为 s3p://
+ 用于检查点到 s3 的推荐连接器。有关更多信息，请参阅 Apache 文档中[特定于 S3 的内容](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/filesystem/#s3-specific)。
示例 FlinkDeployment 规范：
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: basic-example
spec:
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
state.checkpoints.dir: s3p:///flink-checkpoint/
```
**读取/写入 S3：Hadoop S3 连接器**
+ 将 S3 方案设置为 `s3://` 或 `s3a://`
+ 用于从 S3 读取和写入文件的推荐连接器（仅限实现 [Flinks Filesystem 接口](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/filesystem/)的 S3 连接器）。
+ 默认在 `flink-conf.yaml` 文件中设置 `fs.s3a.aws.credentials.provider`，即 `com.amazonaws.auth.WebIdentityTokenCredentialsProvider`。如果完全覆盖默认值 `flink-conf`，并且正在与 S3 进行交互，请务必使用此提供程序。
示例 FlinkDeployment 规范
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: basic-example
spec:
job:
jarURI: local:///opt/flink/examples/streaming/WordCount.jar
args: [ "--input", "s3a:///PATH", "--output", "s3a://