Realtime Compute for Apache Flink:Livre blanc sur les performances (test de référence Nexmark)

Étape 1 : Télécharger l'outil de test

Téléchargez et extrayez le package de l'outil de test Nexmark nexmark-flink.tar.gz.

La structure des répertoires après extraction est la suivante :

nexmark-flink/
├── run_nexmark.sh          # Test entry script
├── nexmark_env.sh          # Environment variable configuration file (requires editing)
├── bin/                    # Runtime scripts
├── conf/                   # Flink job configurations
├── lib/                    # JAR files (to be uploaded to the console)
└── queries-vvp/            # Nexmark Query SQL files

Étape 2 : Charger le fichier JAR Nexmark

1. Connectez-vous à la console Realtime Compute for Apache Flink.

2. Cliquez sur l'espace projet cible. Dans le volet de navigation de gauche, sélectionnez Artifacts > Upload Artifact.

3. Sélectionnez et chargez le fichier nexmark-flink-0.2-SNAPSHOT.jar. Ce fichier se trouve dans le répertoire nexmark-flink/lib de l'outil de test.

4. Une fois le chargement terminé, cliquez sur le nom du fichier pour copier son adresse OSS. Vous aurez besoin de cette adresse lors d'une étape de configuration ultérieure. Le format du chemin d'accès au fichier varie selon le type de stockage :

   • Stockage OSS Bucket : oss://<OSS Bucket name>/artifacts/namespaces/<project space name>/<file name>

     Exemple : oss://oss-test/artifacts/namespaces/flink-default/nexmark-flink-0.2-SNAPSHOT.jar

   • Stockage entièrement géré : oss://flink-fullymanaged-<workspace ID>/artifacts/namespaces/<project space name>/<file name>

     Exemple : oss://flink-fullymanaged-e6a123456789/artifacts/namespaces/flink-default/nexmark-flink-0.2-SNAPSHOT.jar

   Pour afficher le type de stockage de votre espace de travail, accédez à la console de gestion Realtime Compute for Apache Flink, localisez l'espace de travail cible et cliquez sur Actions dans la colonne Details.

Étape 3 : Configurer les paramètres d'exécution

Modifiez le fichier nexmark-flink/nexmark_env.sh et définissez les paramètres suivants.

Étape 4 : Exécuter le test

1. Dans le répertoire nexmark-flink, exécutez la commande suivante.

   ./run_nexmark.sh

2. L'outil de test crée et exécute automatiquement les jobs Nexmark via l'OpenAPI.

3. Une fois le test terminé, la durée de chaque requête s'affiche en millisecondes. L'exemple ci-dessous montre un résultat type :

   INFO  com.github.nexmark.flink.vvp.Nexmark - q0 13078
   ============================================================================
   ✓ Benchmark execution completed successfully
   ============================================================================

Préparation de l'environnement

Créez un cluster Flink à l'aide d'EMR sur ECS avec la configuration suivante :

• Version EMR : EMR-5.21.0

• Spécifications matérielles : trois instances ecs.g6a.xlarge (4 vCPU / 16 Go), comprenant un nœud maître et deux nœuds core.

• Activez les services Hadoop et HDFS.

• Configurez la connexion sans mot de passe entre tous les nœuds. Par exemple, téléchargez votre fichier de clé privée (tel que key.pem) sur le nœud maître. Ajoutez ensuite la configuration suivante au fichier ~/.ssh/config sur le nœud maître. Remplacez les adresses IP et le chemin d'accès au fichier par vos valeurs réelles.

  Host 192.168.0.0
      HostName 192.168.0.0
      User root
      IdentityFile /path/to/key.pem
      StrictHostKeyChecking no
  Host 192.168.0.1
      HostName 192.168.0.1
      User root
      IdentityFile /path/to/key.pem
      StrictHostKeyChecking no
  Host 192.168.0.2
      HostName 192.168.0.2
      User root
      IdentityFile /path/to/key.pem
      StrictHostKeyChecking no

  Utilisez ssh pour vérifier que la connexion sans mot de passe entre les nœuds fonctionne correctement. Si une erreur « bad permissions » se produit, exécutez chmod 600 /path/to/key.pem pour corriger les autorisations.

Préparation logicielle

1. Téléchargez le package Flink cible (Apache Flink Downloads) et le package de test Nexmark (nexmark-flink.tgz). Chargez les packages sur le nœud maître et extrayez-les.

   tar -zxvf flink-1.20.4-bin-scala_2.12.tgz
   tar -zxvf nexmark-flink.tgz
   mv flink-1.20.4 flink
   mv nexmark-flink nexmark

2. Copiez les fichiers JAR du répertoire nexmark/lib vers flink/lib. Ces JAR contiennent le générateur de données Nexmark.

   cp nexmark/lib/* flink/lib/

3. Définissez les variables d'environnement. Modifiez ~/.bashrc, ajoutez la configuration suivante, puis exécutez source ~/.bashrc pour appliquer les modifications.

   Configurez les chemins d'accès en fonction de votre environnement réel.

   export JAVA_HOME=/etc/alternatives/java_sdk_11
   export PATH=$JAVA_HOME/bin:$PATHexport FLINK_HOME=/mnt/disk1/flink
   export HADOOP_CLASSPATH=$(/opt/apps/HADOOP-COMMON/hadoop-common-current/bin/hadoop classpath)

Configuration et démarrage du cluster

1. Configurez les workers Flink. Ce test utilise huit TaskManagers, déployés comme suit : deux sur le nœud maître et trois sur chaque nœud core.

   Modifiez flink/conf/workers, en veillant à remplacer l'adresse IP par la valeur réelle.

   192.168.0.0
   192.168.0.0
   192.168.0.1
   192.168.0.1
   192.168.0.1
   192.168.0.2
   192.168.0.2
   192.168.0.2

2. Remplacez flink/conf/config.yaml par nexmark/conf/config.yaml, et mettez à jour les éléments de configuration suivants :

   • jobmanager.rpc.address : L'adresse IP du nœud maître, par exemple 192.168.0.0.

   • state.checkpoints.dir : Le chemin HDFS, par exemple hdfs:///checkpoints

   • taskmanager.memory.process.size : 4G

3. Modifiez nexmark/conf/nexmark.yaml et définissez nexmark.metric.reporter.host sur l'adresse IP du nœud maître.

4. Distribuez les répertoires flink et nexmark ainsi que les configurations des variables d'environnement vers chaque nœud core.

   Remplacez les adresses IP par vos valeurs réelles.

   scp -r flink 192.168.0.1:/mnt/disk1/
   scp -r flink 192.168.0.2:/mnt/disk1/
   scp -r nexmark 192.168.0.1:/mnt/disk1/
   scp -r nexmark 192.168.0.2:/mnt/disk1/
   scp ~/.bashrc 192.168.0.1:~/
   scp ~/.bashrc 192.168.0.2:~/

   Une fois la distribution terminée, exécutez source ~/.bashrc sur chaque nœud core pour activer les variables d'environnement.

5. Sur le nœud maître, démarrez le cluster Flink.

   flink/bin/start-cluster.sh

6. Initialisez l'environnement de test Nexmark. Ce script configure le Metric Reporter requis sur chaque nœud.

   nexmark/bin/setup_cluster.sh

Définir les limites de ressources

Après le démarrage du cluster Flink, vous devez utiliser cgroups pour limiter l'utilisation du CPU de chaque processus TaskManager à 75 %. Cela empêche les TaskManagers de perdre la connexion en raison de délais d'expiration des signaux de maintien de liaison (heartbeat) causés par la contention des ressources.

Exécutez les commandes suivantes sur tous les nœuds où des TaskManagers sont en cours d'exécution, y compris le nœud maître :

yum install -y libcgroup libcgroup-tools
cgcreate -t root:root -a root:root -g cpu,memory:mygroup
echo 100000 > /sys/fs/cgroup/cpu/mygroup/cpu.cfs_period_us
echo 300000 > /sys/fs/cgroup/cpu/mygroup/cpu.cfs_quota_us
echo $((12 * 1024 * 1024 * 1024)) > /sys/fs/cgroup/memory/mygroup/memory.limit_in_bytes
jps | grep TaskManagerRunner | awk '{print $1}' | xargs cgclassify -g cpu,memory:mygroup

Ici, cpu.cfs_quota_us / cpu.cfs_period_us = 300000 / 100000 = 3, ce qui signifie que le cgroup peut utiliser un maximum de 3 cœurs CPU (75 % de 4 vCPU).

Exécuter Nexmark

Sur le nœud maître, exécutez la commande suivante et attendez qu'elle se termine pour afficher les résultats :

nexmark/bin/run_query.sh q0,q1,q2,q3,q4,q5,q7,q8,q9,q10,q11,q12,q14,q15,q16,q17,q18,q19,q20,q21,q22