本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
# 使用 Amazon EMR on EKS 執行 Flink 作業
Amazon EMR 6.13.0 版及更高版本支援具有 Apache Flink 的 Amazon EMR on EKS 或 Flink Kubernetes Operator，作為 Amazon EMR on EKS 的作業提交模型。使用具有 Apache Flink 的 Amazon EMR on EKS，您可以在自己的 Amazon EKS 叢集上使用 Amazon EMR 發行執行期來部署和管理 Flink 應用程式。在 Amazon EKS 叢集中部署 Flink Kubernetes Operator 之後，即可直接向 Operator 提交 Flink 應用程式。Operator 可管理 Flink 應用程式的生命週期。
**Topics**
+ [設定和使用 Flink Kubernetes Operator](jobruns-flink-kubernetes-operator.md)
+ [使用 Flink 原生 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)
+ [具有 Apache Flink 的 Amazon EMR on EKS 的支援版本](jobruns-flink-security-release-versions.md)
# 設定和使用 Flink Kubernetes Operator
以下頁面描述了如何設定和使用 Flink Kubernetes Operator，以使用 Amazon EMR on EKS 執行 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 通訊的命令列工具。
+ **[安裝 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)**。
+ 建立操作員執行角色。此為選擇性步驟。可以對 Flink 作業和操作員使用相同的角色。如果想要為操作員使用不同的 IAM 角色，可以建立單獨的角色。
+ 更新操作員執行角色的信任政策。必須針對想要用於 Amazon EMR Flink Kubernetes 操作員服務帳戶的角色明確新增一個信任政策項目。可以遵循以下範例格式：
------
#### [ 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 運算子
請使用下列步驟來安裝 Kubernetes Operator for Apache Flink。
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 Chart。
```
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. 等待部署完成，然後驗證 Chart 安裝。
```
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 版本相對應。如需詳細資訊，請參閱[具有 Apache Flink 的 Amazon EMR on EKS 的支援版本](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 運算子
若要升級 Flink 運算子的版本，請遵循下列步驟：
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 及更高版本時，您可以在 Amazon EMR on EKS 上使用 Flink Kubernetes Operator 在應用程式模式下執行 Flink 應用程式。使用 Amazon EMR 6.15.0 及更高版本時，您也可以在工作階段模式下執行 Flink 應用程式。本頁面介紹了可用於透過 Amazon EMR on EKS 執行 Flink 應用程式的兩種方法。
**Topics**
**注意**
提交 Flink 作業時，必須建立 Amazon S3 儲存貯體來儲存高可用性中繼資料。如果不想使用此功能，可以停用它。依預設會啟用此功能。
**必要條件**：在使用 Flink Kubernetes Operator 執行 Flink 應用程式之前，請完成 [針對 Amazon EMR on EKS 設定 Flink Kubernetes Operator](jobruns-flink-kubernetes-operator-setup.md) 和 [安裝 Kubernetes 運算子](jobruns-flink-kubernetes-operator-getting-started.md#jobruns-flink-kubernetes-operator-getting-started-install-operator) 中的步驟。
------
#### [ Application mode ]
使用 Amazon EMR 6.13.0 及更高版本時，您可以在 Amazon EMR on EKS 上使用 Flink Kubernetes Operator 在應用程式模式下執行 Flink 應用程式。
1. 建立`FlinkDeployment`定義檔案，`basic-example-app-cluster.yaml`如下列範例所示。如果您啟用並使用其中一個[選擇加入 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 Operator 提交應用程式至 Flink 的詳細資訊，請參閱 GitHub 上 `apache/flink-kubernetes-operator` 資料夾中的 [Flink Kubernetes Operator 範例](https://github.com/apache/flink-kubernetes-operator/tree/main/examples)。
------
#### [ Session mode ]
使用 Amazon EMR 6.15.0 及更高版本時，您可以在 Amazon EMR on EKS 上使用 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 並執行 Flink 作業，必須建立兩個 Kubernetes 角色：一個 Operator 和一個作業角色。Amazon EMR 會在您安裝 Operator 時預設建立兩個角色。
## Operator 角色
我們使用 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 使用作業角色來建立和管理每個作業的 TaskManagers 和 ConfigMaps。
```
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 原生 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)
+ [Native 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)。
+ 對於您要在其中執行 Flink 應用程式的 [Amazon EKS 叢集建立或更新 Kubeconfig 檔案](https://docs.aws.amazon.com/eks/latest/userguide/create-kubeconfig.html)。
# 開始針對 Amazon EMR on EKS 使用 Flink Native Kubernetes
這些步驟說明如何設定、設定服務帳戶，以及執行 Flink 應用程式。Flink 原生 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 使用者介面。
![\[存取 Flink UI。\]](http://docs.aws.amazon.com/zh_tw/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/)。
# Native Kubernetes 的 Flink JobManager 服務帳戶安全要求
Flink JobManager Pod 使用 Kubernetes 服務帳戶來存取 Kubernetes API 伺服器，以建立和監控 TaskManager Pod。JobManager 服務帳戶必須具有適當的許可，才能建立/刪除 TaskManager Pod，並允許 TaskManager 監看領導者 ConfigMaps，以擷取叢集中 JobManager 和 ResourceManager 的地址。
下列規則適用於此服務帳戶。
```
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 擷取基礎映像
基礎映像包含您存取其他 所需的 Amazon EMR 執行期和連接器 AWS 服務。如果您使用 Amazon EMR on EKS 搭配 Flink 6.14.0 版或更新版本，您可以從 Amazon ECR Public Gallery 取得基礎映像。瀏覽圖庫以尋找映像連結，然後將映像拉到本地工作區。例如，對於 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. 在您的本機工作區建立新的 `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 *Web Dashboard* 將 Flink 與 Amazon Managed Service for Prometheus 整合，以提供任務狀態和指標，或使用監控組態將日誌資料傳送至 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 中擷取指標。Amazon Managed Service for Prometheus 與 Amazon EKS 叢集上已經執行的 Prometheus 伺服器搭配使用。執行 Amazon Managed Service for Prometheus 與 Amazon EMR Flink Operator 的整合將自動部署和設定 Prometheus 伺服器，以便與 Amazon Managed Service for Prometheus 整合。
1. [ 建立 Amazon Managed Service for Prometheus 工作區](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 EMR Flink Operator 與 Amazon Managed Service for Prometheus。
現在您擁有 Amazon Managed Service for Prometheus 工作區、Amazon Managed Service for Prometheus 的專用 IAM 角色以及必要的許可，可以安裝 Amazon EMR Flink Operator。
建立 `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 的運算子中安裝 Prometheus 報告程式。任何未來的 `FlinkDeployment` 也會在 9249 上公開 `metrics` 連接埠。
+ Flink Operator 指標會顯示在標籤 `flink_k8soperator_` 下的 Prometheus 中。
+ Flink Task Manager 指標會顯示在標籤 `flink_taskmanager_` 下的 Prometheus 中。
+ Flink Job Manager 指標會顯示在標籤 `flink_jobmanager_` 下的 Prometheus 中。
# 使用 Flink UI 監控 Flink 任務
若要監控執行中 Flink 應用程式的運作狀態和效能，請使用 *Flink Web Dashboard*。此儀表板提供有關作業狀態、TaskManager 數目以及作業指標和日誌的資訊。它也可讓您檢視和修改 Flink 作業的組態，並與 Flink 叢集互動，以提交或取消作業。
若要存取正在 Kubernetes 上執行的 Flink 應用程式的 Flink Web Dashboard，請執行下列動作：
1. 使用 `kubectl port-forward` 命令將本機連接埠轉送至 Flink 應用程式的 TaskManager Pod 中執行 Flink Web Dashboard 的連接埠。此連接埠預設為 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. 如果您想要在本機使用不同的連接埠，請使用 *local-port*:8081 參數。
```
kubectl port-forward -n flink deployments/basic-example 8080:8081
```
1. 在網頁瀏覽器中，導覽至 `http://localhost:8081` (或 `http://localhost:local-port`，如果您使用自訂本機連接埠) 以存取 Flink Web Dashboard。此儀表板會顯示有關執行中 Flink 應用程式的資訊，例如作業狀態、TaskManager 數目以及作業指標和日誌。
![\[Flink 儀表板 UI 範例\]](http://docs.aws.amazon.com/zh_tw/emr/latest/EMR-on-EKS-DevelopmentGuide/images/sample-flink-dashboard-ui.png)
# 使用監控組態來監控 Flink Kubernetes Operator 和 Flink 任務
監控組態可讓您輕鬆地將 Flink 應用程式和 Operator 日誌的日誌封存設定為 S3 和/或 CloudWatch (您可以選擇其中一個或兩者)。這樣做可將 FluentD 附屬項新增到 JobManager 和 TaskManager Pod，隨後將這些元件的日誌轉發到您設定的接收器。
**注意**
必須為 Flink Oerator 和 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 附屬容器上設定資源限制的組態金鑰。
+ `memoryLimit` (選用) - 預設值為 512Mi。根據需要進行調整。
+ `cpuLimit` (選用) - 此選項沒有預設值。根據需要進行調整。
+ `containerLogRotationConfiguration` (選用) - 控制容器日誌輪換行為。依預設會啟用此功能。
+ `rotationSize` (必要) - 指定日誌輪換的檔案大小。可能的值範圍為 2KB 至 2GB。rotationSize 參數的數值單位部分會以整數形式傳遞。由於不支援小數值，因此可以使用值 1500MB 來指定 1.5GB 的輪換大小。預設值為 2GB。
+ `maxFilesToKeep` (必要) — 指定輪換發生後，要在容器中保留的檔案數上限。下限值是 1，上限值是 50。預設為 10。
## Flink Operator 日誌
也可以使用 Helm Chart 安裝中 `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 附屬容器上設定資源限制的組態金鑰。
+ `memoryLimit` (選用) - 記憶體限制。根據需要進行調整。預設值為 512Mi。
+ `cpuLimit` - CPU 限制。根據需要進行調整。無預設值。
+ `containerLogRotationConfiguration` (選用) - 控制容器日誌輪換行為。依預設會啟用此功能。
+ `rotationSize` (必要) - 指定日誌輪換的檔案大小。可能的值範圍為 2KB 至 2GB。rotationSize 參數的數值單位部分會以整數形式傳遞。由於不支援小數值，因此可以使用值 1500MB 來指定 1.5GB 的輪換大小。預設值為 2GB。
+ `maxFilesToKeep` (必要) — 指定輪換發生後，要在容器中保留的檔案數上限。下限值是 1，上限值是 50。預設為 10。
# Flink 如何支援高可用性和任務彈性
下列各節概述 Flink 如何讓任務更可靠且高可用性。它透過 Flink 高可用性等內建功能執行此操作，並在發生故障時提供各種復原功能。
**Topics**
+ [使用 Flink Operator 和 Flink 應用程式的高可用性 (HA)](jobruns-flink-using-ha.md)
+ [使用 Amazon EMR on EKS 優化 Flink 作業重新啟動時間，以進行任務復原和擴展操作](jobruns-flink-restart.md)
+ [使用 Amazon EMR on EKS 上的 Flink 正常解除委任 Spot 執行個體](jobruns-flink-decommission.md)
# 使用 Flink Operator 和 Flink 應用程式的高可用性 (HA)
本主題說明如何設定高可用性，並說明它在幾個不同的使用案例中的運作方式。這包括當您使用 任務管理員時，以及當您使用 Flink 原生 kubernetes 時。
## Flink Operator 高可用性
我們啟用 Flink Operator 的*高可用性*，以便可以容錯移轉至待命 Flink Operator，在發生故障時將 Operator 控制迴圈中的停機時間降至最低。依預設會啟用「高可用性」，且起始 Operator 複本的預設數目為 2。可以在 `values.yaml` 檔案中設定 Helm Chart 的複本欄位。
下列欄位可自訂：
+ `replicas` (選用，預設值為 2)：將此數字設定為大於 1 可建立其他待命 Operator，並允許更快速地復原作業。
+ `highAvailabilityEnabled` (選用，預設值為 true)：控制是否要啟用 HA。將此參數指定為 true 可啟用多可用區部署支援，並設定正確的 `flink-conf.yaml` 參數。
透過在 `values.yaml` 檔案中設定下列組態，停用 operator 的高可用性。
```
...
imagePullSecrets: []
replicas: 1
# set this to false if you don't want HA
highAvailabilityEnabled: false
...
```
**多可用區部署**
我們會在多個可用區域建立 operator Pod。這是一個軟約束，如果您在不同的可用區域中沒有足夠的資源，將在相同的可用區域中排程您的 operator Pod 。
**確定領導者複本**
如果啟用 HA，則複本使用 Lease 來確定哪些 JM 是領導者，並使用 K8s Lease 進行領導者選舉。您可以描述 Lease 並查看 .Spec.Holder Identity 欄位，以確定目前的領導者
```
kubectl describe lease --lease -n | grep "Holder Identity"
```
**Flink-S3 互動**
**設定存取憑證**
請確定您已設定 IRSA，具有可存取 S3 儲存貯體的適當 IAM 許可。
**從 S3 應用程式模式中擷取作業 jar**
Flink Operator 也支援從 S3 中擷取應用程式 jar。只需在 FlinkDeployment 規格中提供 jarURI 的 S3 位置即可。
也可以使用此功能來下載其他成品，例如 PyFLink 指令碼。生成的 Python 指令碼放在路徑 `/opt/flink/usrlib/` 下。
以下範例示範如何將此功能用於 PyFLink 作業。請注意 jarURI 和引數欄位。
```
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 Flink 文件中的 [S3-specific](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 讀取和寫入檔案的建議連接器 (只有 S3 連接器實作 [Flinks Filesystem 介面](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/filesystem/))。
+ 根據預設，我們在 `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://