E-MapReduce:FAQ

Consulter les journaux du cluster

Résoudre les conflits de packages JAR

Soumettre des jobs depuis une machine externe

Résoudre les noms d'hôte du cluster depuis une machine externe

Vérifier le statut d'un job Flink

• Utilisez la console EMR.

  EMR prend en charge Knox, ce qui vous permet d'accéder aux interfaces web des services tels que YARN et Flink via Internet. Vous pouvez accéder à l'interface web Flink via YARN. Pour plus d'informations, consultez Afficher le statut du job sur l'interface web.

• Utilisez un tunnel SSH. Pour plus d'informations, consultez Créer un tunnel SSH pour accéder aux interfaces web des composants open source.

• Accédez directement à l'API REST 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=***"

  Remarque

  Assurez-vous que votre groupe de sécurité autorise l'accès aux ports 8443 et 8088 pour atteindre l'API REST YARN. Alternativement, assurez-vous que le cluster DataFlow et votre nœud client se trouvent dans le même Virtual Private Cloud (VPC).

Accéder aux journaux d'un job Flink

• Pour un job en cours d'exécution, vous pouvez accéder à ses journaux via l'interface web Flink.

• Pour un job terminé, vous pouvez consulter ses statistiques sur le Flink HistoryServer ou accéder à ses journaux en exécutant la commande yarn logs -applicationId application_xxxx_yyyy. Les journaux des jobs terminés sont stockés par défaut dans le répertoire hdfs:///tmp/logs/$USERNAME/logs/ du cluster HDFS.

Accéder au Flink HistoryServer

Utiliser les connecteurs commerciaux

Les clusters DataFlow incluent des connecteurs commerciaux pour Hologres, SLS, MaxCompute, DataHub, Elasticsearch, ClickHouse, etc. Vous pouvez les utiliser conjointement avec les connecteurs open source. L'exemple suivant montre comment utiliser le connecteur Hologres.

• Développement du job

  1. Téléchargez le package JAR du connecteur commercial depuis le cluster DataFlow (situé dans le répertoire /opt/apps/FLINK/flink-current/opt/connectors). Ensuite, installez le connecteur dans votre environnement Maven local en exécutant la commande suivante.

     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. Ajoutez la dépendance suivante au fichier pom.xml de votre projet.

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

• Exécution du job

  • Méthode 1 :

    1. Copiez le connecteur Hologres dans un répertoire distinct.

       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. Lors de la soumission du job, ajoutez le paramètre suivant à la commande.

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

  • Méthode 2 :

    1. Copiez le connecteur Hologres dans le répertoire /opt/apps/FLINK/flink-current/opt/connectors/ververica-connector-hologres-1.13-vvr-4.0.7.jar sur le client de soumission de job. Cette structure de répertoire doit correspondre à celle du cluster DataFlow.

    2. Lors de la soumission du job, ajoutez le paramètre suivant à la commande.

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

  • Méthode 3 : Intégrez le connecteur Hologres au package JAR de votre job.

Utiliser GeminiStateBackend

Les clusters DataFlow utilisent par défaut le GeminiStateBackend de niveau entreprise, qui offre des performances 3 à 5 fois supérieures à celles des backends d'état open source. Pour plus d'informations sur les configurations avancées de GeminiStateBackend, consultez Configurations du backend d'état de niveau entreprise.

Utiliser un backend d'état open source

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

Consulter les journaux du client

La variable d'environnement FLINK_LOG_DIR spécifie le répertoire des journaux du client Flink. Sa valeur par défaut est /var/log/taihao-apps/flink (avant la version 3.43.0, la valeur par défaut était /mnt/disk1/log/flink). Pour consulter les journaux complets du client, tels que les journaux SQL Client, vérifiez ce répertoire.

Paramètres de job non pris en compte

Lorsque vous exécutez un job Flink depuis la ligne de commande, placez les paramètres du job après le package JAR du job Flink. Par exemple : flink run -d -t yarn-per-job test.jar arg1 arg2.

Résoudre l'erreur « Multiple factories... »

• Cause

  Le classpath contient plusieurs implémentations d'un connecteur. Cela se produit généralement lorsqu'une dépendance de connecteur est empaquetée dans le JAR du job alors que le même connecteur existe également dans le répertoire $FLINK_HOME/lib, provoquant un conflit de dépendances.

• Solution

  La solution consiste à supprimer la dépendance en double. Pour des étapes de dépannage détaillées, consultez Que faire si un package JAR de job entre en conflit avec le package JAR Flink du cluster ?

Activer la haute disponibilité du 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

Consulter les métriques d'un job Flink

1. Dans la console EMR, accédez à la page Monitoring du cluster cible et cliquez sur Metric Monitoring.

2. Dans la liste déroulante Dashboard, sélectionnez FLINK.

3. Sélectionnez l'ID d'application et l'ID de job pour le job que vous souhaitez consulter. Les métriques de surveillance du job s'affichent alors.

   Remarque

   • Les options ID d'application et ID de job sont disponibles uniquement si des jobs Flink sont en cours d'exécution dans le cluster.

   • Certaines métriques, telles que sourceIdleTime, sont générées uniquement si la source et le puits (sink) correspondants sont configurés.

Résoudre les problèmes de connecteurs

Pour les questions courantes concernant le stockage amont et aval, consultez Connecteurs.

Résoudre les erreurs d'accès à OSS sans mot de passe

Gérez le problème en fonction du message d'erreur spécifique :

• Message d'erreur : java.lang.UnsupportedOperationException: Recoverable writers on Hadoop are only supported for HDFS.

  • Cause : Les clusters DataFlow utilisent le JindoSDK intégré pour l'accès sans mot de passe à OSS et des API telles que StreamingFileSink. Aucune configuration supplémentaire issue de la documentation communautaire n'est requise. Son ajout peut provoquer un conflit de dépendances déclenchant cette erreur.

  • Solution : Sur la machine de soumission de job dans votre cluster, vérifiez la présence d'un répertoire oss-fs-hadoop dans le répertoire $FLINK_HOME/plugins. S'il existe, supprimez le répertoire et resoumettez le job.

• Message d'erreur : 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. .....

  • Cause : Dans EMR 3.40 et versions antérieures, les machines du groupe de serveurs maîtres autres que master-1-1 peuvent ne pas disposer des packages JAR liés à Jindo.

  • Solution :

    • Pour EMR 3.40 et versions antérieures : Vérifiez si les packages JAR liés à Jindo, tels que jindo-flink-4.0.0-full.jar, existent dans le répertoire $FLINK_HOME/lib sur la machine de soumission de job. S'ils sont manquants, exécutez la commande suivante pour copier les packages JAR requis dans le répertoire $FLINK_HOME/lib, puis resoumettez le job.

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

    • Pour les versions EMR postérieures à 3.40 :

      • Pour le mode Flink on YARN : Les versions plus récentes disposent d'un mécanisme optimisé pour la prise en charge d'OSS. Les jobs qui lisent et écrivent dans OSS peuvent s'exécuter normalement même si les packages JAR liés à Jindo ne sont pas présents dans le répertoire $FLINK_HOME/lib.

      • Pour les autres modes de déploiement : Vérifiez si les packages JAR liés à Jindo, tels que jindo-flink-4.0.0-full.jar, existent dans le répertoire $FLINK_HOME/lib de la machine de soumission de job. S'ils sont manquants, exécutez la commande suivante pour les copier dans le répertoire $FLINK_HOME/lib, puis resoumettez le job.

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

Résoudre l'erreur « TaskManager heartbeat timed out »

• Cause

  Un délai d'attente du heartbeat d'un TaskManager peut résulter de plusieurs causes. Vérifiez les journaux du TaskManager pour obtenir des messages d'erreur spécifiques. Les causes courantes incluent une insuffisance de mémoire heap ou une erreur de mémoire insuffisante (OOM) due à une fuite de mémoire dans le code du job. Pour plus d'informations, consultez Comment résoudre l'erreur « java.lang.OutOfMemoryError: GC overhead limit exceeded » ?.

• Solution

  Si vous rencontrez cette erreur, augmentez l'allocation de mémoire ou analysez l'utilisation de la mémoire du job pour diagnostiquer davantage le problème.

Résoudre l'erreur « GC overhead limit exceeded »

• Cause

  Le garbage collector (GC) consomme un temps excessif car la mémoire est insuffisante. Les causes courantes incluent une fuite de mémoire dans le code (comme dans une UDF) ou une allocation de mémoire trop faible par rapport aux besoins du job.

• Solution

  • Avant de relancer le job, spécifiez le paramètre JVM suivant à l'aide de l'indicateur -D pour enregistrer un vidage de tas (heap dump) lorsqu'une OutOfMemoryError se produit : -D env.java.opts="-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp/dump.hprof".

  • Ajoutez le paramètre env.java.opts: -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp/dump.hprof au fichier flink-conf.yaml pour configurer les vidages de tas en cas d'OutOfMemoryError.

  Une fois l'erreur reproduite, vous pouvez analyser le fichier de vidage de tas spécifié par HeapDumpPath à l'aide d'outils tels que MAT ou jvisualvm pour déterminer la cause racine.

Zéro « Records Received » pour les jobs à opérateur unique

Ceci est normal. La métrique Records Received de Flink décrit la communication de données entre différents opérateurs. Lorsqu'un job est optimisé en un seul opérateur, cette métrique sera toujours égale à 0.

Activer les graphiques en flammes (flame graphs) pour les jobs Flink

Les graphiques en flammes visualisent la consommation de CPU à travers les méthodes au sein d'un processus, ce qui vous aide à identifier les goulets d'étranglement de performance. Flink prend en charge les graphiques en flammes depuis la version 1.13, mais la fonctionnalité est désactivée par défaut pour éviter d'affecter les jobs de production. Si vous devez utiliser les graphiques en flammes pour analyser les performances d'un job, accédez à l'onglet Configure du service Flink dans la console EMR. Dans le fichier flink-conf.yaml, ajoutez un nouvel élément de configuration avec le paramètre rest.flamegraph.enabled et définissez sa valeur sur true. Pour obtenir des instructions sur l'ajout d'un élément de configuration, consultez Gérer les éléments de configuration.

Pour plus d'informations sur les graphiques en flammes, consultez Flame Graphs.

Résoudre l'erreur « NoSuchFieldError: DEPLOYMENT_MODE »

• Cause

  Le package JAR de votre job inclut directement ou indirectement une dépendance flink-core incompatible avec la version Flink du cluster, provoquant un conflit de dépendances.

• Solution

  Ajoutez la configuration suivante à votre fichier pom.xml pour définir le scope de la dépendance flink-core sur provided. Cela résout le problème.

  <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>

  Remarque

  Vous devez modifier la version pour correspondre à votre version Flink.

  Pour localiser plus précisément la source de cette dépendance, consultez Que faire si un package JAR de job entre en conflit avec le package JAR Flink du cluster ?.