本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
# 使用 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://