Realtime Compute for Apache Flink:Informe técnico de rendimiento (benchmark Nexmark)

Paso 1: Descargar la herramienta de pruebas

Descargue y extraiga el paquete de la herramienta de pruebas Nexmark nexmark-flink.tar.gz.

La estructura de directorios después de la extracción es la siguiente:

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

Paso 2: Cargar el JAR de Nexmark

1. Inicie sesión en la Realtime Compute for Apache Flink console.

2. Haga clic en el project space deseado. En la barra de navegación izquierda, seleccione Artifacts > Upload Artifact.

3. Seleccione y cargue el archivo nexmark-flink-0.2-SNAPSHOT.jar. El archivo se encuentra en el directorio nexmark-flink/lib de la herramienta de pruebas.

4. Una vez completada la carga, haga clic en el nombre del archivo para copiar su dirección de OSS. Necesitará esta dirección para un paso de configuración posterior. El formato de la ruta del archivo varía según el tipo de almacenamiento:

   • Almacenamiento en bucket de OSS: oss://<OSS Bucket name>/artifacts/namespaces/<project space name>/<file name>

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

   • Almacenamiento totalmente administrado: oss://flink-fullymanaged-<workspace ID>/artifacts/namespaces/<project space name>/<file name>

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

   Para consultar el tipo de almacenamiento de su workspace, vaya a la consola de administración de Realtime Compute for Apache Flink, busque el workspace correspondiente y haga clic en Actions en la columna Details.

Paso 3: Configurar los parámetros de ejecución

Edite el archivo nexmark-flink/nexmark_env.sh y establezca los siguientes parámetros.

Paso 4: Ejecutar la prueba

1. En el directorio nexmark-flink, ejecute el siguiente comando.

   ./run_nexmark.sh

2. La herramienta de pruebas crea y ejecuta automáticamente los jobs de Nexmark mediante OpenAPI.

3. Al finalizar la prueba, se muestra la duración de cada consulta en milisegundos. El siguiente ejemplo muestra una salida de muestra:

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

Preparación del entorno

Cree un clúster de Flink mediante EMR en ECS con la siguiente configuración:

• Versión de EMR: EMR-5.21.0

• Especificación de hardware: tres instancias ecs.g6a.xlarge (4 vCPU / 16 GiB), incluido un nodo maestro y dos nodos core.

• Habilite los servicios Hadoop y HDFS.

• Configure el inicio de sesión sin contraseña entre todos los nodos. Por ejemplo, cargue su archivo de clave privada (como key.pem) en el nodo maestro. Luego, agregue la siguiente configuración al archivo ~/.ssh/config del nodo maestro. Reemplace las direcciones IP y la ruta del archivo con sus valores reales.

  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

  Use ssh para comprobar que el inicio de sesión sin contraseña entre los nodos funciona correctamente. Si ocurre un error de "bad permissions", ejecute chmod 600 /path/to/key.pem para corregir los permisos.

Preparación del software

1. Descargue el paquete de Flink deseado (Apache Flink Downloads) y el paquete de pruebas Nexmark (nexmark-flink.tgz). Cargue los paquetes en el nodo maestro y extráigalos.

   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. Copie los archivos JAR del directorio nexmark/lib a flink/lib. Estos JAR contienen el generador de datos de Nexmark.

   cp nexmark/lib/* flink/lib/

3. Establezca las variables de entorno. Edite ~/.bashrc, agregue la siguiente configuración y luego ejecute source ~/.bashrc para que los cambios surtan efecto.

   Configure las rutas según su entorno real.

   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)

Configuración e inicio del clúster

1. Configure los workers de Flink. Esta prueba utiliza ocho TaskManagers, distribuidos de la siguiente manera: dos en el nodo maestro y tres en cada nodo core.

   Edite flink/conf/workers y asegúrese de reemplazar la dirección IP con el valor real.

   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. Reemplace flink/conf/config.yaml por nexmark/conf/config.yaml y actualice los siguientes elementos de configuración:

   • jobmanager.rpc.address: la dirección IP del nodo maestro, por ejemplo, 192.168.0.0.

   • state.checkpoints.dir: la ruta de HDFS, por ejemplo, hdfs:///checkpoints

   • taskmanager.memory.process.size: 4G

3. Edite nexmark/conf/nexmark.yaml y establezca nexmark.metric.reporter.host en la dirección IP del nodo maestro.

4. Distribuya los directorios flink y nexmark, así como las configuraciones de variables de entorno, a cada nodo core.

   Reemplace las direcciones IP con sus valores reales.

   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:~/

   Una vez completada la distribución, ejecute source ~/.bashrc en cada nodo core para que las variables de entorno surtan efecto.

5. En el nodo maestro, inicie el clúster de Flink.

   flink/bin/start-cluster.sh

6. Inicialice el entorno de pruebas de Nexmark. Este script configura el Metric Reporter necesario en cada nodo.

   nexmark/bin/setup_cluster.sh

Establecer límites de recursos

Después de iniciar el clúster de Flink, debe usar cgroups para limitar el uso de CPU de cada proceso TaskManager al 75%. Esto evita que los TaskManagers pierdan la conexión debido a tiempos de espera de heartbeat causados por contención de recursos.

Ejecute los siguientes comandos en todos los nodos donde se ejecuten TaskManagers, incluido el nodo maestro:

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

Aquí, cpu.cfs_quota_us / cpu.cfs_period_us = 300000 / 100000 = 3, lo que significa que el cgroup puede usar un máximo de 3 núcleos de CPU (75% de 4 vCPU).

Ejecutar Nexmark

En el nodo maestro, ejecute el siguiente comando y espere a que finalice para ver los resultados:

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