配置Spark Native引擎
配置场景
Spark Native引擎是通过使用向量化的C++加速库，实现对Spark算子性能加速的一种技术方案。传统的SparkSQL是基于行式数据，通过JVM的codegen来实现查询加速的，由于JVM对生成的java代码存在各种约束，比如方法长度，参数个数等，以及行式数据对内存带宽的利用率不足，因此存在性能提升空间。使用成熟的向量化的c++加速库后，数据采用向量化格式存在内存中，可以提高带宽利用率，并通过批量的数处理获得加速效果。
通过开启Spark Native引擎特性，获得SparkSQL的性能加速。配置Spark Native引擎后不支持Spark大SQL查询和删除。
使用约束
- 本章节仅适用于MRS 3.6.0.1及之后版本。
- 支持spark-sql以及spark-beeline任务。
- 数据类型
Spark 类型
Substrait
Velox
计算
Scan
写入
备注
BooleanType
bool
BOOLEAN
✅
✅
✅
-
ByteType
i8
TINYINT
✅
❌
✅
ParquetScan fallback
ShortType
i16
SMALLINT
✅
✅
✅
-
IntegerType
i32
INTEGER
✅
✅
✅
-
LongType
i64
BIGINT
✅
✅
✅
-
FloatType
fp32
REAL
✅
✅
✅
-
DoubleType
fp64
DOUBLE
✅
✅
✅
-
StringType
string
VARCHAR
✅
✅
✅
-
BinaryType
binary
VARBINARY
✅
✅
✅
-
DateType
date
DATE
✅
✅
✅
-
TimestampType
timestamp
TIMESTAMP
✅
⚠️
⚠️
Parquet仅 INT96+dict；ORC scan/write不支持
DecimalType(p,s)
decimal
DECIMAL/SHORT_DECIMAL/HUGEINT
✅
✅
✅
按精度选底层表示
NullType
nothing
UNKNOWN
✅
✅
✅
-
CharType(n)
fixedChar
VARCHAR
✅
⚠️
✅
ORC scan强制 fallback
YearMonthIntervalType
intervalYear
INTERVAL_YEAR_MONTH
✅
✅
❌
写入 Parquet/ORC/CSV 不支持
ArrayType
list
ARRAY
✅
⚠️
❌
element为 Struct/Array时scan fallback；native write不支持
MapType
map
MAP
✅
⚠️
❌
key为Struct或value 为Array时scan fallback；native write不支持
StructType
struct
ROW
✅
⚠️
❌
Parquet/CSV写入不支持；native write 不支持
DayTimeIntervalType
-
-
❌
❌
❌
不支持
CalendarIntervalType
-
-
❌
❌
❌
不支持
UUIDType
-
-
❌
❌
❌
不支持
VarcharType(n)
-
-
❌
❌
❌
不支持
TimestampNTZType
-
-
❌
❌
❌
不支持
- 数据格式
名称
读取
写入
Parquet
√
√
ORC
√
√
CSV/Text
√
x
Iceberg（Parquet/ORC）
√
√
- 算子Fallback回Spark场景
算子
说明
SubqueryExec / SubqueryBroadcastExec
子查询保持行式执行
AggregateInPandasExec
Pandas聚合
MapInPandasExec
PandasMap
FlatMapGroupsInPandasExec
Pandas FlatMapGroups
WindowInPandasExec
Pandas窗口
LocalTableScanExec
本地表扫描（数据量极小，无需原生加速）
CommandExec 系列
DDL/DML命令（Show/Describe/Create/Alter 等）
QueryStageExec
AQE阶段边界节点
CustomShuffleReaderExec
AQE Shuffle读取
CreateDataSourceTableAsSelectCommand
CTAS始终回退
InsertIntoHadoopFsRelationCommand（bucket）
有bucket时不支持
- 表达式Fallback回spark场景
表达式
限制
from_json
不支持enablePartialResults=false
from_json
不支持options参数
from_json
不支持大小写敏感（caseSensitive=true）
from_json
不支持schema中重复key
from_json
不支持corrupt record类型列
to_json
不支持options参数
数组/高阶函数
表达式
限制
array_filter + 带index的lambda
不支持，
array_forall + 带index的lambda
不支持，与array_filter限制相同
array_exists + 带index的lambda
不支持，与array_filter限制相同
array_transform + 带index的 ambda
不支持，与array_filter限制相同
其他表达式
表达式
限制
str_to_map
仅支持mapKeyDedupPolicy=EXCEPTION
unbase64 + failOnError=true
不支持
TryEval / ANSI 算术
仅支持整型（Long/Int/Short/Byte）
PreciseTimestampConversion 涉及 TimestampNTZType
不支持
HiveHash
不支持
Cast 从 ArrayType 到 String
通过Rewrite规则间接支持
base64
需启用chunkBase64String
- 文件系统
名称
是否支持
OBS
√
HDFS
√
Pacific
√
- 平台
名称
是否支持
X86、ARM、X86与ARM混合部署
√
- 使用方式
名称
是否支持
spark-sql yarn模式
√
spark-sql local模式
√
spark beeline
√
- 数据类型限制
不支持的类型
影响范围
TimestampNTZType
所有含此类型的算子
DayTimeIntervalType
不在schema白名单
UserDefinedType
不支持自定义类型
- Native Shuffle(spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager)，支持的压缩算法(spark.io.compression.codec)。
名称
是否支持
lz4
√
lzf
x
snappy
x
zstd
√
gzip
x
bzip2
x
服务端配置参数
- 在Manager系统中，选择“集群 > 服务 > Spark > 配置”，单击“全部配置”，选择“Spark（服务） > Native”，配置如下参数：
参数
说明
默认值
spark.plugins
Spark用到的插件，参数值设置为org.apache.gluten.GlutenPlugin。
说明：如果已经配置了spark.plugins，则可以将
org.apache.gluten.GlutenPlugin加到其中，用逗号","隔开。
空
spark.gluten.memory.dynamic.offHeap.sizing.enabled
是否开启Spark Native统一内存管理。设置为true，Native加速需要使用统一内存管理。
true
spark.shuffle.manager
shuffle管理器。
选择org.apache.spark.shuffle.mass.MassShuffleManager开启memartsstore远程shuffle服务。
选择org.apache.spark.shuffle.sort.ColumnarShuffleManager开启native engine列式shuffle服务。
选择org.apache.spark.shuffle.gluten.mass.MassShuffleManager开启native engine列式memartsstore远程shuffle服务
sort
spark.sql.orc.impl
当需要开启Native读写ORC表时，需要设置为native。
hive
- 单击“保存”，保存已经修改的配置，并重启过期的实例。
- 选择“概览 > 更多 > 下载客户端”，安装并使用该客户端。
客户端配置参数
- 如果已经按照服务端配置参数配置，并下载安装新的客户端，可跳过该章节。
- 在Spark客户端的“{客户端安装目录}/Spark/spark/conf/spark-defaults.conf”配置文件中进行设置，修改如下参数：
参数
说明
默认值
spark.plugins
Spark用到的插件，参数值设置为org.apache.gluten.GlutenPlugin。
说明：如果已经配置了spark.plugins，则可以将org.apache.gluten.GlutenPlugin加到其中，用逗号","隔开。
空
spark.gluten.memory.dynamic.offHeap.sizing.enabled
是否开启Spark Native统一内存管理。设置为true，Native加速需要使用统一内存管理。
true
spark.shuffle.manager
shuffle管理器。
选择org.apache.spark.shuffle.mass.MassShuffleManager开启memartsstore远程shuffle服务。
选择org.apache.spark.shuffle.sort.ColumnarShuffleManager开启native engine列式shuffle服务。
选择org.apache.spark.shuffle.gluten.mass.MassShuffleManager开启native engine列式memartsstore远程shuffle服务
r
sort
spark.sql.orc.impl
当需要开启Native读写ORC表时，需要设置为native。
hive
- 在客户端目录下执行“source bigdata_env”刷新环境变量
使用说明
可通过检查执行计划的方式验证Native Engine是否开启，执行如下语句：
spark-sql -e "explain select * from database.data_source"
当执行计划中出现CHNativeColumnarToRow字样，即可认为Native Engine已经开启。
数据一致性校验工具
数据一致性验证工具基于Shell脚本，通过调用Spark-shell启动Scala应用程序。该Scala程序实现了核心的比较逻辑，对比Spark执行查询和通过Spark Native方式执行查询时生成的表数据。Spark和Spark Native的查询结果
- 按照客户端配置参数开启Spark Native。
- 进入“{客户端安装目录}/Spark/spark/tool”目录，执行sh md5compare.sh --help可查看使用方法：
md5compare.sh --dbname <name> --query-path <path> --spark_tbl_location <path> --native_tbl_location <path> --spark_tbl_name <name> --native_tbl_name <name> --clear_tbl_data <boolean> --driver_mem <MEM> --executor_mem <MEM> --executor_num <NUM> --executor_core <NUM>
表1 参数说明 参数
是否必填
参数说明
--dbname <数据库名>
必填
执行 SQL 查询时使用的目标数据库
--query-path < 文件路径 >
必填
SQL 查询脚本的路径，该文件仅支持单条SQL
--spark_tbl_location < 目录路径 >
可选
Spark 查询结果的外部表存储目录，默认值为 /tmp/spark
--native_tbl_location < 目录路径 >
可选
原生查询结果的外部表存储目录，默认值为 /tmp/native
--spark_tbl_name < 表名 >
可选
存储 Spark 查询结果的临时表名称，默认值为 spark_data
--native_tbl_name < 表名 >
可选
存储原生查询结果的临时表名称，默认值为 native_data
--clear_tbl_data < 布尔值 >
可选
对比结束后是否删除临时表及外部表存储路径，默认值为 true
--driver_mem < 内存值 >
可选
Driver 进程使用的内存大小，默认值为 4G
--executor_mem < 内存值 >
可选
每个 Executor 进程使用的内存大小，默认值为 2G
--executor_num < 数值 >
可选
启动的 Executor 进程总数量，默认值为 4
--executor_core < 数值 >
可选
为每个 Executor 进程分配的 CPU 核心数，默认值为 1
- 执行示例：
sh md5compare.sh --dbname tpcds_hive_spark2x_2 --query-path /opt/query/q1.sql --spark_tbl_location /tmp/spark --native_tbl_location /tmp/native --spark_tbl_name spark_data --native_tbl_name native_data --clear_tbl_data true --driver_mem 8G --executor_mem 4G --executor_num 2 --executor_core 2
针对Spark和Native Engine在处理decimal，float以及double时可能存在的小数部分的差异，一致性校验工具统一取到小数点后一位做比较。例如:3.14159，一致性工具在比较时取3.1做比较。
差异说明
开启native engine后，执行结果和spark runtime存在细微差异：
- 由于spark runtime和native engine在处理decimal的舍入规则不同，造成decimal结果最后一位可能存在差异。
- 读取json格式的binary类型数据时，由于Spark在写入时进行了base64编码：如 '0101' 写入json文件后为'MDEwMQ=='，Spark runtime读取会base64解码，但native engine读取时未base64解码，需要unbase64()函数解码。
- Native Engine不支持SET TIME ZONE INTERVAL语法。
常见问题
- 服务端配置： 登录Manager页面，选择“集群 > 服务 > Spark > 配置”，单击“全部配置”，选择“Spark（服务） > Native”，配置如下参数：
参数
说明
默认值
spark.gluten.sql.columnar.backend.ch.shuffle.preferSpill
当shuffle内存达到门限时，是否溢写到磁盘
修改配置为true
false
spark.gluten.sql.columnar.backend.ch.spillThreshold
shuffle内存门限值, 当shuffle内存超过该值，会溢写磁盘
建议设置为128M
0
spark.gluten.sql.columnar.maxBatchSize
shuffle算子一个批次处理的行数, 影响Spark Native引擎中shuffle算子的内存占用
可设置为1024
4096
- 客户端配置：
在Spark客户端节点的“{客户端安装目录}/Spark/spark/conf/spark-defaults.conf”配置文件中进行设置，新增如下参数：
spark.gluten.sql.columnar.backend.ch.shuffle.preferSpill=true spark.gluten.sql.columnar.backend.ch.spillThreshold=128M spark.gluten.sql.columnar.maxBatchSize=1024
问题二：执行SQL任务时，如果Fallback Node过多，性能比Spark更差，可考虑使用动态开关关闭Spark native引擎。
set spark.gluten.enabled=false;
调优指南
- 文件被切分成多个分片后，最大分片的大小（默认64MB)：
spark.gluten.sql.columnar.backend.ch.file.split.size=268435456
- 内存超过executor限制是否抛出异常（设置为false可允许native超限使用内存以提升性能，但有被OOM killer结束executor进程的风险）。
spark.gluten.sql.columnar.backend.ch.throwIfMemoryExceed = false
- 是否允许var_sample, corvar_sample函数的精度损失：
spark.gluten.sql.columnar.precision.loss.allowed = true
- 是否开启decimal溢出检查，关闭可提升decimal计算性能：
spark.gluten.sql.columnar.backend.ch.runtime_settings.decimal_check_overflow=false
即时编译
spark.gluten.sql.columnar.backend.ch.runtime_config.compile_expressions=1
