Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.
# In che modo Flink supporta l'elevata disponibilità e la resilienza del lavoro
--lease -n | grep "Holder Identity"
```
**Interazione Flink-S3**
**Configurazione delle credenziali di accesso**
Assicurati di aver configurato IRSA con le autorizzazioni IAM appropriate per accedere al bucket S3.
**Recupero dei jar dei processi dalla modalità di applicazione S3**
L'operatore Flink supporta anche il recupero dei jar delle applicazioni da S3. È sufficiente fornire la posizione S3 per il JARuri nelle specifiche. FlinkDeployment
Puoi anche usare questa funzione per scaricare altri artefatti come gli script. PyFlink Lo script Python risultante viene inserito nel percorso `/opt/flink/usrlib/`.
L'esempio seguente mostra come utilizzare questa funzionalità per un lavoro. PyFlink Osserva i campi jarURI e 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
```
**Connettori S3 di Flink**
Flink viene fornito con due connettori S3 (elencati di seguito). Nelle sezioni seguenti viene spiegato quando utilizzare un determinato connettore.
**Creazione di checkpoint: connettore Presto S3**
+ Imposta lo schema S3 su s3p://
+ Il connettore consigliato da utilizzare per la creazione di checkpoint su s3. Per ulteriori informazioni, consulta [S3-specific](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/filesystem/#s3-specific) nella documentazione di Apache Flink.
Specificazione di esempio: FlinkDeployment
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: basic-example
spec:
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
state.checkpoints.dir: s3p:///flink-checkpoint/
```
**Lettura e scrittura su S3: connettore Hadoop S3**
+ Imposta lo schema S3 su `s3://` o (`s3a://`)
+ Il connettore consigliato per la lettura e la scrittura di file da S3 (unico connettore S3 a implementare l'[interfaccia FileSystem di Flink](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/filesystem/)).
+ Per impostazione predefinita, abbiamo impostato `fs.s3a.aws.credentials.provider` il `flink-conf.yaml` file, che è. `com.amazonaws.auth.WebIdentityTokenCredentialsProvider` Se ignori completamente il valore `flink-conf` predefinito e stai interagendo con S3, assicurati di utilizzare questo provider.
FlinkDeployment Specifiche di esempio
```
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" ]
parallelism: 2
upgradeMode: stateless
```
## Flink Job Manager
"
emrReleaseLabel: "emr-6.13.0-flink-latest"
jobManager:
resource:
memory: "2048m"
cpu: 1
replicas: 2
highAvailabilityEnabled: true
storageDir: "s3://"
taskManager:
resource:
memory: "2048m"
cpu: 1
```
Di seguito sono riportate le descrizioni delle configurazioni HA di cui sopra in Job Manager (definite in .spec.jobManager):
+ `highAvailabilityEnabled` (facoltativa, l'impostazione predefinita è true): impostala su `false ` se non desideri abilitare l'HA e non desideri utilizzare le configurazioni HA fornite. Puoi comunque manipolare il campo "repliche" per configurare manualmente l'HA.
+ `replicas`(facoltativo, l'impostazione predefinita è 2): l'impostazione di questo numero su un valore maggiore di 1 crea altri standby JobManagers e consente un ripristino più rapido del lavoro. Se disabiliti l'HA, devi impostare il numero di repliche su 1, altrimenti continuerai a ricevere errori di convalida (è supportata solo 1 replica se l'HA non è abilitata).
+ `storageDir` (obbligatoria): poiché per impostazione predefinita si utilizza 2 come numero di repliche, è necessario fornire una storageDir persistente. Attualmente questo campo accetta solo percorsi S3 come posizione di archiviazione.
**Località dei pod**
Se abiliti HA, cerchiamo anche di collocare i pod nella stessa AZ, il che comporta un miglioramento delle prestazioni (latenza di rete ridotta grazie alla presenza di pod nella stessa). AZs Si tratta di una procedura basata sul migliore tentativo, il che significa che se non disponi di risorse sufficienti nella zona in cui è programmata la maggior parte dei tuoi pod, i pod rimanenti verranno comunque programmati, ma potrebbero finire su un nodo esterno a tale AZ.
**Determinazione della replica leader**
Se HA è abilitato, le repliche utilizzano un lease per determinare quale delle due JMs è la leader e utilizzano una K8s Configmap come archivio dati per archiviare questi metadati. Se vuoi determinare il leader, puoi verificare il contenuto della Configmap e la chiave `org.apache.flink.k8s.leader.restserver` sotto i dati per trovare il pod K8s con l'indirizzo IP. Inoltre, puoi utilizzare i seguenti comandi bash.
```
ip=$(kubectl get configmap -n -cluster-config-map -o json | jq -r ".data[\"org.apache.flink.k8s.leader.restserver\"]" | awk -F: '{print $2}' | awk -F '/' '{print $3}')
kubectl get pods -n NAMESPACE -o json | jq -r ".items[] | select(.status.podIP == \"$ip\") | .metadata.name"
```
## Processo Flink: Kubernetes nativo