BoostKit 大数据 OmniRuntime 性能优化原理分析-171主机测评

1. 背景

Apache Spark 是目前主流的大数据计算框架。随着数据量的增长，基于 JVM 的 Spark 原生执行引擎在性能上面临以下瓶颈：

CPU 流水线效率低：Spark 采用“火山模型”（Volcano Model），以行（Row）为单位处理数据。每次处理一行数据涉及虚函数调用（Next()），导致 CPU 分支预测失败和指令缓存未命中（Instruction Cache Miss）增加。

GC（垃圾回收）开销：大量数据对象在 JVM 堆（Heap）中创建和销毁，引发频繁的 GC，严重时导致 OOM。

硬件利用率不足：JVM 的 JIT 编译器在利用 SIMD（单指令多数据）等现代 CPU 特性方面，效率不如 C++ 编译器。

华为鲲鹏 BoostKit 大数据使能套件提供了 OmniRuntime（OmniOperator），通过将 Spark 的计算密集型算子下沉到 Native C++ 层，并利用鲲鹏处理器的 NEON 指令集进行矢量化加速，解决了上述问题。

华为鲲鹏 BoostKit官网：https://www.hikunpeng.com/boostkit。

2. 核心原理

BoostKit OmniRuntime 并不是对 Spark 做“局部打补丁式优化”，而是从执行模型、内存模型和数据布局三个维度，对 Spark SQL 物理执行层进行重构式加速。

OmniRuntime 是 BoostKit 大数据加速底座的核心组件，包含 OmniOperator（算子加速）、OmniShuffle（Shuffle 加速）等模块，本章将重点讲解其对 Spark SQL 的执行模型、内存模型、数据布局的重构式优化原理。

其优化核心可以概括为三个关键词：

Native 化、矢量化、列式化

从 Spark SQL 的执行链路来看，OmniRuntime 主要作用于 Catalyst 生成物理计划之后、Task 实际执行之前，将 JVM Row-based 执行路径，透明替换为 Native Columnar 执行路径。

2.1 Native 矢量化计算

在原生 Spark SQL 执行模型中，算子运行在 JVM 上，数据以 Row 为单位逐行处理，即使启用了 WholeStageCodeGen，本质仍然是标量执行模型。这种模式在 CPU 指令级并行、Cache 利用率以及函数调用开销方面均存在天然瓶颈。

// Spark Row-based 标量执行（简化示例）
while (iter.hasNext()) {
UnsafeRow row = (UnsafeRow) iter.next();
int a = row.getInt(0);
int b = row.getInt(1);
output.append(a + b);
}

OmniRuntime 在物理执行阶段接管算子执行，将输入数据按照 Batch 组织（通常为 4096 行），并以列式结构传入 C++ Native Runtime。在 Native 层，算子逻辑不再以“行”为中心，而是以“列”为基本计算单元，充分利用鲲鹏 CPU 的 ARM NEON SIMD 指令集进行并行计算。

// C++ Native + NEON SIMD 计算示例
int32_t* col_a = ptr_a;
int32_t* col_b = ptr_b;
int32_t* col_out = ptr_out;

for (int i = 0; i < row_count; i += 4) {
int32x4_t vec_a = vld1q_s32(col_a + i);
int32x4_t vec_b = vld1q_s32(col_b + i);

int32x4_t vec_res = vaddq_s32(vec_a, vec_b);

vst1q_s32(col_out + i, vec_res);
}

通过 SIMD，一条指令即可同时完成多个数据元素的计算，相比 JVM 标量执行，显著提升了 CPU 吞吐能力。同时，列式数据在内存中连续存放，访问模式更加规整，进一步提高了 L1/L2 Cache 的命中率，为算子执行带来叠加收益。

2.2 Off-Heap 内存管理

在传统 Spark 执行过程中，大量中间结果存储在 JVM 堆内，容易引发频繁 GC，尤其在 Join、Aggregate 等场景下，GC 抖动往往成为性能瓶颈。

OmniRuntime 采用 Off-Heap（堆外）内存模型，将算子执行过程中的核心数据全部存储在 Native 内存中。JVM 侧仅保留必要的元信息（如内存地址、长度等），从而彻底规避 GC 对执行路径的影响。

// Native 侧堆外内存分配
void* buffer = aligned_alloc(64, buffer_size);

// 构建列式向量
int32_t* column_data = reinterpret_cast<int32_t*>(buffer);

在 JVM 与 Native 之间，OmniRuntime 通过 JNI 传递内存地址，实现 Zero-Copy 数据共享。Spark 的 UnsafeRow 或 ColumnarBatch 不再发生 Java 堆与 Native 堆之间的数据拷贝，Native Runtime 可以直接对数据进行读写。

// JVM 侧：传递 Native 地址
long nativeAddr = omniColumn.getNativeAddress();
nativeExecute(nativeAddr, rowCount);

// Native 侧：直接访问 JVM 传递的地址
int32_t* data = reinterpret_cast<int32_t*>(nativeAddr);

这种 Off-Heap + Zero-Copy 的设计，使 OmniRuntime 在大数据量、高并发场景下依然能够保持稳定的延迟表现，并显著降低 Executor 对 JVM Heap 大小的依赖。

Heap 分配代码：

public class HeapGcTest {

public static void main(String[] args) throws Exception {
long last = System.nanoTime();

for (int i = 0; i < 200; i++) {
// 模拟 Join / Aggregate 产生的大量中间结果
byte[] data = new byte[50 * 1024 * 1024]; // 50MB

long now = System.nanoTime();
System.out.printf("Iter %d latency: %.2f ms%n",
i, (now – last) / 1_000_000.0);
last = now;

Thread.sleep(20);
}
}
}

运行的时候需要加运行参数：

-Xms512m -Xmx512m -XX:+PrintGCDetails -XX:+PrintGCTimeStamps

Off-Heap 分配：

import java.nio.ByteBuffer;

public class OffHeapGcTest {

public static void main(String[] args) throws Exception {
long last = System.nanoTime();

for (int i = 0; i < 200; i++) {
// 使用 Off-Heap 内存
ByteBuffer buffer =
ByteBuffer.allocateDirect(50 * 1024 * 1024);

long now = System.nanoTime();
System.out.printf("Iter %d latency: %.2f ms%n",
i, (now – last) / 1_000_000.0);
last = now;

Thread.sleep(20);
}
}
}

运行结果对比：

表格还在加载中，请等待加载完成后再尝试复制

2.3 Columnar Shuffle

Shuffle 是 Spark SQL 执行过程中最容易成为瓶颈的阶段之一。传统 Shuffle 需要在列式与行式之间反复转换，并依赖 Java 序列化与反序列化，带来较高的 CPU 和内存开销。

OmniRuntime 在 Shuffle 阶段引入 Columnar Shuffle 机制，直接对列式数据进行序列化与网络传输，避免了“列转行 → 编解码 → 行转列”的冗余过程。

// Columnar Shuffle 序列化示例（简化）
for (auto& column : batch.columns()) {
serialize(column.data(), column.size());
}

在数据压缩方面，OmniRuntime 使用 Native 实现的 LZ4 / ZSTD 算法，相比 Java 版本具有更低的指令开销和更高的压缩吞吐。

// Native ZSTD 压缩示例
size_t compressed_size = ZSTD_compress(
dst, dst_capacity,
src, src_size,
compression_level
);

通过列式传输与 Native 压缩的组合优化，Shuffle 阶段的 CPU 开销和网络传输成本均得到有效降低，在 Join、Group By 等典型 OLAP 场景中表现尤为明显。

3. 实战配置

华为鲲鹏 BoostKit OmniRuntime 为 Apache Spark 提供了开箱即用的性能优化能力，仅需完成简单的环境部署与配置，即可让 Spark SQL 作业自动启用 Native 列式执行引擎，无需修改业务代码。

在官网我们也可以找到相对应的安装教程：

软件包安装：

可通过鲲鹏官网提供的 YUM 源或离线包安装 BoostKit 软件包：

# 配置鲲鹏 YUM 源（以 openEuler 为例）
echo -e "[kunpeng-boostkit]\\nname=Kunpeng BoostKit – openEuler 22.03 LTS\\nbaseurl=https://repo.oepkgs.net/openeuler/rpm/openEuler-22.03-LTS/extras/aarch64/\\nenabled=1\\ngpgcheck=0" > /etc/yum.repos.d/kunpeng-boostkit.repo

# 安装 OmniOperator 软件包
yuminstall-yboostkit-omnioperator–nogpgcheck

Spark 配置：

OmniRuntime 的配置支持两种方式：全局配置（spark-defaults.conf）（对所有作业生效）或作业提交时指定（仅对当前作业生效），可根据业务需求选择。

全局配置：

编辑 Spark 集群的 spark-defaults.conf 文件（默认路径：$SPARK_HOME/conf/spark-defaults.conf），添加以下配置项：

# 1. 启用 OmniOperator 列式执行扩展（核心配置）
spark.sql.extensions=com.huawei.boostkit.spark.ColumnarPlugin

# 2. 配置 Native 动态库路径（指向 BoostKit 安装的 .so 库目录）
spark.driver.extraLibraryPath=/opt/omni-operator/lib
spark.executor.extraLibraryPath=/opt/omni-operator/lib

# 3. 启用列式 Shuffle（优化 Shuffle 性能，建议开启）
spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager

# 4. 优化内存配置：减少堆内内存，增加堆外内存供 Native 层使用
spark.executor.memory=20g # 堆内内存
spark.memory.offHeap.enabled=true # 开启堆外内存
spark.memory.offHeap.size=20g # 堆外内存大小（建议与堆内内存相当）

# 5. 可选：关闭 Spark 原生 WholeStageCodeGen（OmniRuntime 已提供更优的 Native 代码生成）
spark.sql.codegen.wholeStage=false

作业级配置：

提交 Spark 作业时，通过 –conf 参数指定配置项，示例如下：

spark-submit \\
–class com.example.YourSparkJob \\
–master yarn \\
–deploy-mode cluster \\
–executor-memory 20g \\
–num-executors 30 \\
–executor-cores 8 \\
# OmniRuntime 核心配置
–conf spark.sql.extensions=com.huawei.boostkit.spark.ColumnarPlugin \\
–conf spark.driver.extraLibraryPath=/opt/omni-operator/lib \\
–conf spark.executor.extraLibraryPath=/opt/omni-operator/lib \\
–conf spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager \\
–conf spark.memory.offHeap.enabled=true \\
–conf spark.memory.offHeap.size=20g \\
# 作业 Jar 包
your-spark-job.jar

验证配置生效：

在 Spark Shell 中执行 SQL 并通过 explain() 方法查看物理执行计划，若计划中出现Columnar 前缀的算子（如 ColumnarHashAggregate、ColumnarFileScan），则表示 Native 列式执行引擎已生效。

# 启动 Spark Scala Shell（需带上 OmniRuntime 配置）
spark-shell \\
–conf spark.sql.extensions=com.huawei.boostkit.spark.ColumnarPlugin \\
–conf spark.driver.extraLibraryPath=/opt/omni-operator/lib

# 执行 SQL 并查看执行计划
scala> spark.sql("SELECT sum(price) FROM sales GROUP BY city").explain()

配置成功：

4. 性能对比

为验证鲲鹏 BoostKit OmniRuntime 对 Spark 的性能提升效果，我们基于鲲鹏 920 集群，采用 TPC-DS 1TB 标准数据集，设计了对比测试方案，通过实战代码执行测试并量化分析性能差异。

4.1 测试环境

本次测试的硬件与软件环境保持统一，仅区分是否启用 OmniRuntime 优化，保证其他条件都是一致的，这样我们可以更加直观的看出鲲鹏 BoostKit OmniRuntime 对 Spark 的性能提升效果：

表格还在加载中，请等待加载完成后再尝试复制

4.2 实战测试代码

我们通过 Spark Submit 提交 TPC-DS 测试任务，分别执行开源 Spark 和 BoostKit Spark 测试，以下是核心测试代码与提交脚本：

数据读取与查询执行代码（Scala）：

import org.apache.spark.sql.SparkSession
import java.util.concurrent.TimeUnit

object TPCDSBenchmark {
def main(args: Array[String]): Unit = {
// 初始化 SparkSession
val spark = SparkSession.builder()
.appName("TPCDS-Benchmark")
.getOrCreate()

// 读取 TPC-DS 1TB 数据集（Parquet 格式，已提前生成）
val tpcdsPath = "hdfs:///tpcds/1tb/parquet"
// 示例：读取核心表（实际测试会遍历所有 25 张表）
val storeSalesDF = spark.read.parquet(s"$tpcdsPath/store_sales")
val dateDimDF = spark.read.parquet(s"$tpcdsPath/date_dim")
storeSalesDF.createOrReplaceTempView("store_sales")
dateDimDF.createOrReplaceTempView("date_dim")

// 执行查询并统计耗时（多次执行取平均值，排除缓存影响）
val repeatTimes = 5
val times = new Array[Long](repeatTimes)
for (i <- 0 until repeatTimes) {
val start = System.nanoTime()
spark.sql(q67).collect() // 触发执行
val end = System.nanoTime()
times(i) = TimeUnit.NANOSECONDS.toSeconds(end – start)
println(s"第 ${i+1} 次执行耗时：${times(i)} 秒")
}

// 计算平均耗时
val avgTime = times.sum / repeatTimes.toDouble
println(s"\\nQ67 平均执行耗时：$avgTime 秒")

spark.stop()
}
}

开源 Spark 提交脚本（spark-submit-open.sh）：

#!/bin/bash
spark-submit \\
–class TPCDSBenchmark \\
–master yarn \\
–deploy-mode cluster \\
–executor-memory 20g \\
–driver-memory 8g \\
–num-executors 60 \\
–executor-cores 8 \\
# 启用 OmniRuntime 核心配置
–conf spark.sql.extensions=com.huawei.boostkit.spark.ColumnarPlugin \\
–conf spark.driver.extraLibraryPath=/opt/omni-operator/lib \\
–conf spark.executor.extraLibraryPath=/opt/omni-operator/lib \\
–conf spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager \\
–conf spark.memory.offHeap.enabled=true \\
–conf spark.memory.offHeap.size=20g \\
–conf spark.sql.adaptive.enabled=true \\
./tpcds-benchmark-1.0.jar

BoostKit Spark 提交脚本（spark-submit-boostkit.sh）：

代码运行结果对比：

4.3数据对比

通过执行上述测试代码，我们统计了整体查询耗时、典型复杂查询耗时、CPU 利用率等核心指标，具体对比数据如下表：

表格还在加载中，请等待加载完成后再尝试复制

从测试数据可以看出：

BoostKit Spark 借助 Native 矢量化计算、Off-Heap 内存管理和 Columnar Shuffle 优化，整体性能较开源 Spark 提升近 50%，复杂查询性能直接翻倍。

CPU 利用率的提升和 GC 停顿时间的降低，说明 OmniRuntime 有效解决了 JVM 行式执行的瓶颈，让硬件资源得到更充分的利用。

Shuffle 阶段的耗时大幅减少，印证了 Columnar Shuffle 规避数据格式冗余转换的优化效果。

5. 总结

BoostKit OmniRuntime 通过将 SIMD 和 Native 内存管理技术引入 Spark，解决了 JVM 带来的性能瓶颈。用户只需简单的配置即可获得 Native 级的性能体验，从而在不增加硬件成本的情况下提升大数据计算效率。

BoostKit 大数据 OmniRuntime 性能优化原理分析

1. 背景