性能白皮书（Nexmark性能测试）

步骤一：下载测试工具

下载 Nexmark 测试工具压缩包nexmark-flink.tar.gz并解压。

解压后的目录结构如下：

nexmark-flink/
├── run_nexmark.sh          # 测试入口脚本
├── nexmark_env.sh          # 环境变量配置文件（需编辑）
├── bin/                    # 运行脚本
├── conf/                   # Flink 作业配置
├── lib/                    # JAR 包（需上传至控制台）
└── queries-vvp/            # Nexmark Query SQL 文件

步骤二：上传 Nexmark JAR

1. 登录实时计算控制台。

2. 单击进入目标项目空间，在左侧导航栏文件管理 > 上传资源。

3. 选择并上传 nexmark-flink-0.2-SNAPSHOT.jar 文件。该文件位于测试工具的 nexmark-flink/lib 目录下。

4. 上传完成后，单击目标文件名称复制 OSS 地址。该地址在后续配置参数时使用。文件路径格式因存储类型而异：

   • OSS Bucket 存储：oss://<OSS Bucket 名称>/artifacts/namespaces/<项目空间名称>/<文件名>

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

   • 全托管存储：oss://flink-fullymanaged-<工作空间ID>/artifacts/namespaces/<项目空间名称>/<文件名>

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

   如需查看工作空间的存储类型，在实时计算管理控制台单击目标工作空间操作列下的详情查看。

步骤三：配置运行参数

编辑 nexmark-flink/nexmark_env.sh 文件，填写以下参数。

步骤四：运行测试

1. 在 nexmark-flink 目录下执行以下命令。

   ./run_nexmark.sh

2. 测试工具通过 OpenAPI 自动创建并运行 Nexmark 作业。

3. 运行完成后，输出各 Query 的运行时长（毫秒）。示例如下：

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

环境准备

通过 EMR on ECS 创建 Flink 集群，集群配置如下：

• EMR 版本：EMR-5.21.0

• 硬件规格：3 台 ecs.g6a.xlarge（4 vCPU / 16 GiB），1 台 Master + 2 台 Core

• 开启 Hadoop 和 HDFS 服务

• 配置好各个节点之间的免密登录。例如，将安全组的私钥 key.pem上传到 master 节点，然后在 master 节点的 ~/.ssh/config 写入如下配置，注意将 IP 和文件路径替换为实际值：

  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

  使用 ssh 验证各节点间免密登录是否正常。若报错 "bad permissions"，执行 chmod 600 /path/to/key.pem 修正权限。

软件准备

1. 下载目标版本的 Flink 包（Apache Flink Downloads）和 Nexmark 测试包（nexmark-flink.tgz），上传至 Master 节点并解压。

   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. 将 nexmark/lib 目录下的 JAR 文件复制到 flink/lib，这些 JAR 包含 Nexmark 数据生成器。

   cp nexmark/lib/* flink/lib/

3. 设置环境变量。编辑 ~/.bashrc，加入以下配置后执行 source ~/.bashrc 使其生效。

   根据实际环境配置相关路径。

   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)

配置并启动 Flink 集群

1. 配置 Flink Workers。本文使用 8 个 TaskManager，采用"Master 节点 2 个 + 每个 Core 节点各 3 个"的部署方式。

   编辑 flink/conf/workers，注意将IP替换为实际值。

   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. 用 nexmark/conf/config.yaml 替换 flink/conf/config.yaml，并更新以下配置项：

   • jobmanager.rpc.address：Master 节点 IP，如 192.168.0.0

   • state.checkpoints.dir：HDFS 路径，如 hdfs:///checkpoints

   • taskmanager.memory.process.size：4G

3. 编辑 nexmark/conf/nexmark.yaml，将 nexmark.metric.reporter.host 设为 Master 节点 IP。

4. 将 flink 和 nexmark 目录及环境变量配置分发到各 Core 节点。

   注意将IP替换为实际值。

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

   分发完成后，在各 Core 节点执行 source ~/.bashrc 使环境变量生效。

5. 在 Master 节点启动 Flink 集群。

   flink/bin/start-cluster.sh

6. 初始化 Nexmark 测试环境。该脚本会在各节点上配置 Nexmark 运行所需的 Metric Reporter。

   nexmark/bin/setup_cluster.sh

水位设置

Flink 集群启动后，需通过 cgroup 将各 TaskManager 进程的 CPU 水位限制在 75% 以内，避免因资源争抢导致 TaskManager 心跳超时失联。

在所有运行 TaskManager 的节点上（含 Master 节点）执行以下命令：

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

其中 cpu.cfs_quota_us / cpu.cfs_period_us = 300000 / 100000 = 3，表示该 cgroup 最多使用 3 个 CPU 核心（4 vCPU 的 75%）。

运行 Nexmark

在 Master 节点执行以下命令，等待运行结束后查看结果：

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