Performance white paper (Nexmark benchmark)

Step 1: Download the test tool

Download and extract the Nexmark test tool package nexmark-flink.tar.gz.

The directory structure after extraction is as follows:

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

Step 2: Upload the Nexmark JAR

1. Log in to the Realtime Compute for Apache Flink console.

2. Click the target project space. In the left navigation bar, select File Management > Upload Resource.

3. Select and upload the nexmark-flink-0.2-SNAPSHOT.jar file. The file is located in the nexmark-flink/lib directory of the test tool.

4. After the upload is complete, click the file name to copy its OSS address. You will need this address for a later configuration step. The file path format varies by storage type:

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

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

   • Fully managed storage: oss://flink-fullymanaged-<workspace ID>/artifacts/namespaces/<project space name>/<file name>

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

   To view the storage type of your workspace, go to the Realtime Compute for Apache Flink management console, find the target workspace, and click Details in the Actions column.

Step 3: Configure runtime parameters

Edit the nexmark-flink/nexmark_env.sh file and set the following parameters.

Step 4: Run the test

1. In the nexmark-flink directory, run the following command.

   ./run_nexmark.sh

2. The test tool automatically creates and runs the Nexmark jobs using OpenAPI.

3. After the test is complete, the duration for each query is displayed in milliseconds. The following example shows a sample output:

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

Environment preparation

Create a Flink cluster using EMR on ECS with the following configuration:

• EMR version: EMR-5.21.0

• Hardware specification: Three ecs.g6a.xlarge instances (4 vCPU / 16 GiB), including one master node and two core nodes.

• Enable the Hadoop and HDFS services.

• Configure passwordless login between all nodes. For example, upload your private key file (such as key.pem) to the master node. Then, add the following configuration to the ~/.ssh/config file on the master node. Replace the IP addresses and file path with your actual values.

  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 to verify that passwordless login between the nodes is working correctly. If a "bad permissions" error occurs, run chmod 600 /path/to/key.pem to correct the permissions.

Software preparation

1. Download the target Flink package (Apache Flink Downloads) and the Nexmark test package (nexmark-flink.tgz). Upload the packages to the master node and extract them.

   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. Copy the JAR files in the nexmark/lib directory to flink/lib. These JARs contain the Nexmark data generator.

   cp nexmark/lib/* flink/lib/

3. Set the environment variables. Edit ~/.bashrc, add the following configuration, and then run source ~/.bashrc for the changes to take effect.

   Configure the paths based on your actual environment.

   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)

Cluster configuration and startup

1. Configure Flink workers. This test uses eight TaskManagers, deployed as follows: two on the master node and three on each core node.

   Edit flink/conf/workers, making sure to replace the IP address with the actual value.

   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. Replace flink/conf/config.yaml with nexmark/conf/config.yaml, and update the following configuration items:

   • jobmanager.rpc.address: The IP address of the master node, such as 192.168.0.0.

   • state.checkpoints.dir: The HDFS path, such as hdfs:///checkpoints

   • taskmanager.memory.process.size: 4G

3. Edit nexmark/conf/nexmark.yaml and set nexmark.metric.reporter.host to the IP address of the master node.

4. Distribute the flink and nexmark directories and environment variable configurations to each core node.

   Replace the IP addresses with your actual values.

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

   After the distribution is complete, run source ~/.bashrc on each core node to make the environment variables take effect.

5. On the master node, start the Flink cluster.

   flink/bin/start-cluster.sh

6. Initialize the Nexmark test environment. This script configures the required Metric Reporter on each node.

   nexmark/bin/setup_cluster.sh

Set resource limits

After the Flink cluster starts, you must use cgroups to limit the CPU usage of each TaskManager process to 75%. This prevents TaskManagers from losing connection due to heartbeat timeouts caused by resource contention.

Run the following commands on all nodes where TaskManagers are running, including the master node:

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

Here, cpu.cfs_quota_us / cpu.cfs_period_us = 300000 / 100000 = 3, which means the cgroup can use a maximum of 3 CPU cores (75% of 4 vCPUs).

Run Nexmark

On the master node, run the following command and wait for it to complete to view the results:

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