Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.
# Cómo apoya Flink la alta disponibilidad y la resiliencia al trabajo
--lease -n | grep "Holder Identity"
```
**Interacción entre Flink y S3**
**Configuración de las credenciales de acceso**
Asegúrese de haber configurado las IRSA con los permisos de IAM adecuados para acceder al bucket de S3.
**Búsqueda de archivos jar de trabajo desde el modo aplicación de S3**
El operador de Flink también admite la búsqueda de archivos jar de aplicaciones desde S3. Solo tiene que proporcionar la ubicación S3 del JaRuri en su especificación. FlinkDeployment
También puedes usar esta función para descargar otros artefactos, como PyFlink scripts. El script de Python resultante se coloca en la ruta `/opt/flink/usrlib/`.
El siguiente ejemplo muestra cómo utilizar esta función para un PyFlink trabajo. Tenga en cuenta los campos jarURI y 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
```
**Conectores de Flink de S3**
Flink viene empaquetado con dos conectores de S3 (enumerados a continuación). En las siguientes secciones se explica cuándo usar cada conector.
**Punto de control: conector de Presto de S3**
+ Establezca el esquema de S3 en s3p://
+ El conector recomendado para establecer el punto de control en s3. Para obtener más información, consulte [Específico de S3](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/filesystem/#s3-specific) en la documentación de Apache Flink.
Ejemplo de FlinkDeployment especificación:
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: basic-example
spec:
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
state.checkpoints.dir: s3p:///flink-checkpoint/
```
**Lectura y escritura en S3: conector Hadoop S3**
+ Establezca el esquema de S3 en `s3://` o en ( `s3a://` )
+ El conector recomendado para leer y escribir archivos desde S3 (solo el conector de S3 que implementa la interfaz del [sistema de archivos de Flink](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/filesystem/)).
+ De forma predeterminada, establecemos `fs.s3a.aws.credentials.provider` en el archivo `flink-conf.yaml`, que es `com.amazonaws.auth.WebIdentityTokenCredentialsProvider`. Si anula completamente el valor predeterminado `flink-conf` y está interactuando con S3, asegúrese de usar este proveedor.
Ejemplo de FlinkDeployment especificación
```
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 JobManager
"
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
```
Las siguientes son descripciones de las configuraciones de alta disponibilidad anteriores en Job Manager (definidas en .spec.JobManager):
+ `highAvailabilityEnabled` (opcional, el valor predeterminado es verdadero): configúrela en `false ` si no quiere activar la alta disponibilidad y no quiere utilizar las configuraciones de alta disponibilidad proporcionadas. Aún puede manipular el campo “réplicas” para configurar de forma manual la alta disponibilidad.
+ `replicas`(opcional, el valor predeterminado es 2): si se establece este número en mayor que 1, se crea otro modo de espera JobManagers y permite una recuperación más rápida del trabajo. Si deshabilita la alta disponibilidad, debe establecer el recuento de réplicas en 1 o seguirá recibiendo errores de validación (solo se admite 1 réplica si la alta disponibilidad no está habilitada).
+ `storageDir` (obligatorio): dado que utilizamos el recuento de réplicas como 2 de forma predeterminada, debemos proporcionar un storageDir persistente. Actualmente, este campo solo acepta rutas de S3 como ubicación de almacenamiento.
**Localidad del pod**
Si habilitas la alta disponibilidad, también intentamos colocar los pods en la misma zona de disponibilidad, lo que mejora el rendimiento (se reduce la latencia de la red al tener los pods en la misma zona AZs). Se trata de un proceso de mejor esfuerzo, ya que, si no dispone de recursos suficientes en la zona de disponibilidad en la que están programados la mayoría de sus pods, los demás pods seguirán estando programados, pero es posible que acaben en un nodo fuera de esta zona de disponibilidad.
**Determinar la réplica líder**
Si la alta disponibilidad está habilitada, las réplicas utilizan un arrendamiento para determinar cuál de ellos JMs es el líder y utilizan un mapa de configuración del K8s como almacén de datos para almacenar estos metadatos. Si quiere determinar el líder, consulte el contenido del Configmap y busque en los datos la clave `org.apache.flink.k8s.leader.restserver` para encontrar los pod de K8 con la dirección IP. También puede utilizar los siguientes comandos 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"
```
## Trabajo de Flink: Kubernetes nativo