E-MapReduce:Preguntas frecuentes

Consulta de registros del clúster

Resolución de conflictos de paquetes JAR

Envío de trabajos desde una máquina externa

Resolución de nombres de host del clúster desde una máquina externa

Comprobación del estado de los trabajos de Flink

• Utilice la consola de EMR.

  EMR es compatible con Knox, lo que permite acceder a las interfaces web de servicios como YARN y Flink a través de internet. Puede acceder a la interfaz web de Flink mediante YARN. Para obtener más información, consulte View the job status on the web UI.

• Utilice un túnel SSH. Para obtener más información, consulte Create an SSH tunnel to access web UIs of open source components.

• Acceda directamente a la API REST de YARN.

  curl --compressed -v  -H "Accept: application/json" -X GET "http://master-1-1:8088/ws/v1/cluster/apps?states=RUNNING&queue=default&user.name=***"

  Nota

  Asegúrese de que su security group permita el acceso a los puertos 8443 y 8088 para alcanzar la API REST de YARN. Como alternativa, asegúrese de que el clúster DataFlow y su nodo cliente se encuentren en la misma Virtual Private Cloud (VPC).

Acceso a los registros de trabajos de Flink

• Para un trabajo en ejecución, puede acceder a sus registros a través de la interfaz web de Flink.

• Para un trabajo finalizado, puede consultar sus estadísticas en Flink HistoryServer o acceder a sus registros mediante la ejecución del comando yarn logs -applicationId application_xxxx_yyyy. Los registros de los trabajos finalizados se almacenan de forma predeterminada en el directorio hdfs:///tmp/logs/$USERNAME/logs/ del clúster HDFS.

Acceso a Flink HistoryServer

Uso de connectors comerciales

Los clústeres DataFlow incluyen connectors comerciales para Hologres, SLS, MaxCompute, DataHub, Elasticsearch, ClickHouse, entre otros. Puede utilizarlos junto con los connectors de código abierto. El siguiente ejemplo muestra cómo usar el connector de Hologres.

• Desarrollo del trabajo

  1. Descargue el paquete JAR del connector comercial desde el clúster DataFlow (ubicado en el directorio /opt/apps/FLINK/flink-current/opt/connectors). Luego, instale el connector en su entorno Maven local mediante la ejecución del siguiente comando.

     mvn install:install-file -Dfile=/path/to/ververica-connector-hologres-1.13-vvr-4.0.7.jar -DgroupId=com.alibaba.ververica -DartifactId=ververica-connector-hologres -Dversion=1.13-vvr-4.0.7 -Dpackaging=jar

  2. Agregue la siguiente dependencia al archivo pom.xml de su proyecto.

     <dependency>
         <groupId>com.alibaba.ververica</groupId>
         <artifactId>ververica-connector-hologres</artifactId>
         <version>1.13-vvr-4.0.7</version>
         <scope>provided</scope>
     </dependency>

• Ejecución del trabajo

  • Método 1:

    1. Copie el connector de Hologres a un directorio independiente.

       hdfs mkdir hdfs:///flink-current/opt/connectors/hologres/
       hdfs cp hdfs:///flink-current/opt/connectors/ververica-connector-hologres-1.13-vvr-4.0.7.jar  hdfs:///flink-current/opt/connectors/hologres/ververica-connector-hologres-1.13-vvr-4.0.7.jar

    2. Al enviar el trabajo, agregue el siguiente parámetro al comando.

       -D yarn.provided.lib.dirs=hdfs:///flink-current/opt/connectors/hologres/

  • Método 2:

    1. Copie el connector de Hologres al directorio /opt/apps/FLINK/flink-current/opt/connectors/ververica-connector-hologres-1.13-vvr-4.0.7.jar en el cliente de envío de trabajos. Esta estructura de directorios debe coincidir con la del clúster DataFlow.

    2. Al enviar el trabajo, agregue el siguiente parámetro al comando.

       -C file:///opt/apps/FLINK/flink-current/opt/connectors/ververica-connector-hologres-1.13-vvr-4.0.7.jar

  • Método 3: Empaquete el connector de Hologres dentro del paquete JAR de su trabajo.

Uso de GeminiStateBackend

Los clústeres DataFlow utilizan GeminiStateBackend de nivel empresarial de forma predeterminada, el cual ofrece un rendimiento de 3 a 5 veces superior al de los state backends de código abierto. Para obtener más información sobre las configuraciones avanzadas de GeminiStateBackend, consulte Enterprise-grade state backend configurations.

Uso de un state backend de código abierto

flink run-application -t yarn-application -D state.backend=rocksdb  /opt/apps/FLINK/flink-current/examples/streaming/TopSpeedWindowing.jar

Consulta de registros del cliente

La variable de entorno FLINK_LOG_DIR especifica el directorio de registros del cliente de Flink. Su valor predeterminado es /var/log/taihao-apps/flink (antes de la versión 3.43.0, el valor predeterminado era /mnt/disk1/log/flink). Para consultar registros completos del cliente, como los registros de SQL Client, revise este directorio.

Parámetros de trabajo sin efecto

Cuando ejecute un trabajo de Flink desde la línea de comandos, coloque los parámetros del trabajo después del paquete JAR del trabajo de Flink. Por ejemplo: flink run -d -t yarn-per-job test.jar arg1 arg2.

Resolución del error "Multiple factories..."

• Causa

  El classpath contiene múltiples implementaciones de un connector. Esto suele ocurrir cuando una dependencia del connector se empaqueta en el JAR del trabajo mientras que el mismo connector también existe en el directorio $FLINK_HOME/lib, lo que provoca un conflicto de dependencias.

• Solución

  La solución consiste en eliminar la dependencia duplicada. Para obtener pasos detallados de solución de problemas, consulte What do I do if a job JAR package conflicts with the cluster's Flink JAR package?

Habilitación de alta disponibilidad para JobManager

high-availability: zookeeper
high-availability.zookeeper.quorum: 192.168.**.**:2181,192.168.**.**:2181,192.168.**.**:2181
high-availability.zookeeper.path.root: /flink
high-availability.storageDir: hdfs:///flink/recovery

Visualización de métricas de trabajos de Flink

1. En la consola de EMR, vaya a la página Monitoring del clúster objetivo y haga clic en Metric Monitoring.

2. En la lista desplegable Dashboard, seleccione FLINK.

3. Seleccione el ID de aplicación y el ID de trabajo correspondientes al trabajo que desea consultar. A continuación, aparecerán las métricas de monitoreo del trabajo.

   Nota

   • Las opciones de ID de aplicación e ID de trabajo solo están disponibles si hay trabajos de Flink ejecutándose en el clúster.

   • Algunas métricas, como sourceIdleTime, solo se generan si se configuran el source y el sink correspondientes.

Solución de problemas de connectors

Para preguntas frecuentes sobre almacenamiento ascendente y descendente, consulte Connectors.

Resolución de errores en el acceso a OSS sin contraseña

Aborde el problema según el mensaje de error específico:

• Mensaje de error: java.lang.UnsupportedOperationException: Recoverable writers on Hadoop are only supported for HDFS.

  • Causa: Los clústeres DataFlow utilizan JindoSDK integrado para el acceso a OSS sin contraseña y API como StreamingFileSink. No se requiere configuración adicional de la documentación de la comunidad. Agregarla puede provocar un conflicto de dependencias que desencadene este error.

  • Solución: En la máquina de envío de trabajos de su clúster, compruebe si existe un directorio oss-fs-hadoop dentro del directorio $FLINK_HOME/plugins. Si existe, elimine el directorio y vuelva a enviar el trabajo.

• Mensaje de error: Could not find a file system implementation for scheme 'oss'. The scheme is directly supported by Flink through the following plugin: flink-oss-fs-hadoop. .....

  • Causa: En EMR 3.40 y versiones anteriores, es posible que falten paquetes JAR relacionados con Jindo en las máquinas del grupo de servidores maestros distintas de master-1-1.

  • Solución:

    • Para EMR 3.40 y versiones anteriores: Compruebe si existen paquetes JAR relacionados con Jindo, como jindo-flink-4.0.0-full.jar, en el directorio $FLINK_HOME/lib de la máquina de envío de trabajos. Si faltan, ejecute el siguiente comando para copiar los paquetes JAR necesarios al directorio $FLINK_HOME/lib y vuelva a enviar el trabajo.

      cp /opt/apps/extra-jars/flink/jindo-flink-*-full.jar $FLINK_HOME/lib

    • Para versiones de EMR posteriores a la 3.40:

      • Para el modo Flink on YARN: Las versiones más recientes cuentan con un mecanismo optimizado para la compatibilidad con OSS. Los trabajos que leen y escriben en OSS pueden ejecutarse correctamente incluso si no hay paquetes JAR relacionados con Jindo en el directorio $FLINK_HOME/lib.

      • Para otros modos de despliegue: Compruebe si existen paquetes JAR relacionados con Jindo, como jindo-flink-4.0.0-full.jar, en el directorio $FLINK_HOME/lib de la máquina de envío de trabajos. Si faltan, ejecute el siguiente comando para copiarlos al directorio $FLINK_HOME/lib y vuelva a enviar el trabajo.

        cp /opt/apps/extra-jars/flink/jindo-flink-*-full.jar $FLINK_HOME/lib

Resolución del error "TaskManager heartbeat timed out"

• Causa

  Un tiempo de espera agotado en el heartbeat del TaskManager puede deberse a varias razones. Revise los registros del TaskManager para buscar mensajes de error específicos. Las causas comunes incluyen memoria heap insuficiente o un error de falta de memoria (OOM) provocado por una fuga de memoria en el código del trabajo. Para obtener más información, consulte How do I resolve the "java.lang.OutOfMemoryError: GC overhead limit exceeded" error?.

• Solución

  Si encuentra este error, aumente la asignación de memoria o analice el uso de memoria del trabajo para diagnosticar mejor el problema.

Resolución del error "GC overhead limit exceeded"

• Causa

  El recolector de basura (GC) consume demasiado tiempo porque la memoria es insuficiente. Las causas comunes incluyen una fuga de memoria en el código (como en una UDF) o una asignación de memoria demasiado baja para los requisitos del trabajo.

• Solución

  • Antes de volver a ejecutar el trabajo, especifique el siguiente parámetro de JVM mediante la bandera -D para guardar un volcado de heap cuando ocurra un OutOfMemoryError: -D env.java.opts="-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp/dump.hprof".

  • Agregue el parámetro env.java.opts: -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp/dump.hprof al archivo flink-conf.yaml para configurar volcados de heap ante errores OutOfMemoryError.

  Una vez que el error vuelva a ocurrir, puede analizar el archivo de volcado de heap especificado por HeapDumpPath mediante herramientas como MAT o jvisualvm para determinar la causa raíz.

Cero "Records Received" en trabajos de operador único

Esto es normal. La métrica Records Received de Flink describe la comunicación de datos entre distintos operadores. Cuando un trabajo se optimiza a un solo operador, esta métrica siempre será 0.

Habilitación de gráficos de llama para trabajos de Flink

Los gráficos de llama visualizan el consumo de CPU entre los métodos de un proceso, lo que ayuda a identificar cuellos de botella de rendimiento. Flink admite gráficos de llama desde la versión 1.13, pero la función está deshabilitada de forma predeterminada para no afectar los trabajos de producción. Si necesita utilizar gráficos de llama para analizar el rendimiento del trabajo, vaya a la pestaña Configure del servicio Flink en la consola de EMR. En el archivo flink-conf.yaml, agregue un nuevo elemento de configuración con el parámetro rest.flamegraph.enabled y establezca su valor en true. Para obtener instrucciones sobre cómo agregar un elemento de configuración, consulte Manage configuration items.

Para obtener más información sobre los gráficos de llama, consulte Flame Graphs.

Resolución del error "NoSuchFieldError: DEPLOYMENT_MODE"

• Causa

  El paquete JAR de su trabajo incluye directa o indirectamente una dependencia flink-core incompatible con la versión de Flink del clúster, lo que provoca un conflicto de dependencias.

• Solución

  Agregue la siguiente configuración a su archivo pom.xml para establecer el scope de la dependencia flink-core en provided. Esto resuelve el problema.

  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-core</artifactId>
    <!-- Change to your own flink version -->
    <version>1.16.1</version>
    <scope>provided</scope>
  </dependency>

  Nota

  Debe cambiar el valor de version por su versión de Flink.

  Para localizar con mayor precisión el origen de esta dependencia, consulte What do I do if a job JAR package conflicts with the cluster's Flink JAR package?.