# 配置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) | - | - | ✅ | ⚠️ | ❌ | 仅Parquet格式支持scan；ORC格式scan强制fallback | | TimestampNTZType | - | - | ❌ | ❌ | ❌ | 不支持 | ![](https://support.huaweicloud.com/cmpntguide-lts-mrs/public_sys-resources/note_3.0-zh-cn.png) 对于ORC格式，当表中存在varchar(n)和char(n)类型，那么整个scan算子会强制fallback回Spark处理。 - 数据格式： | 名称 | 读取 | 写入 | |:---|:---|:---| | 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场景。 JSON 函数： | 表达式 | 限制 | |:---|:---| | 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 | 不支持， - 带 index：不支持，回退Spark。 SELECT filter(array(1, 2, 3), (x, i) -\> i \> 0); - 结果: \[2, 3\]（i 是数组下标，从0开始） - 区别在于lambda函数的参数个数： - (x) -\> ... --- 1 个参数，只用元素值，Velox 支持 - (x, i) -\> ... --- 2 个参数，第二个 i 是元素下标，Velox 不支持 | | array_forall + 带index的lambda | 不支持，与array_filter限制相同 | | array_exists + 带index的lambda | 不支持，与array_filter限制相同 | | array_transform + 带index的lambda | 不支持，与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 | √ | ![](https://support.huaweicloud.com/cmpntguide-lts-mrs/public_sys-resources/note_3.0-zh-cn.png) 跨Region访问OBS时，需要在spark-default.conf中添加如下配置： ``` spark.gluten.velox.obs.access = hdfs spark.gluten.velox.obs.write.sdk.type = hdfs ``` - 支持平台： | 名称 | 是否支持 | |:---|:---| | X86、ARM、X86与ARM混合部署 | √ | ![](https://support.huaweicloud.com/cmpntguide-lts-mrs/public_sys-resources/notice_3.0-zh-cn.png) 部分操作系统（例如SUSE 15系列）没有"/usr/share/zoneinfo/tzdata.zi"文件，会导致时间格式数据报错。需要从其他类型操作系统（例如EulerOS）复制tzdata.zi文件到"/usr/share/zoneinfo/"目录。 - 使用方式： | 名称 | 是否支持 | |:---|:---| | spark-sql yarn模式 | √ | | spark-sql local模式 | √ | | spark beeline | √ | - 数据类型限制： 含以下类型的 schema 会导致算子验证失败并回退： | 不支持的类型 | 影响范围 | |:---|:---| | 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 | #### 服务端配置参数 1. 在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 | 2. 单击"保存"，保存已经修改的配置，并重启过期的实例。 3. 选择"概览 \> 更多 \> 下载客户端"，安装并使用该客户端。 #### 客户端配置参数 1. 如果已经按照[服务端配置参数]配置，并下载安装新的客户端，可跳过该章节。 2. 在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服务。 | sort | | spark.sql.orc.impl | 当需要开启Native读写ORC表时，需要设置为native。 | hive | 3. 在客户端目录下执行"source bigdata_env"刷新环境变量。 #### 使用说明 可通过检查执行计划的方式验证Native Engine是否开启，执行如下语句： **spark-sql -e "explain select \* from database.data_source"** 当执行计划中出现VeloxColumnarToRow字样，即可认为Native Engine已经开启。 ![](https://support.huaweicloud.com/cmpntguide-lts-mrs/zh-cn_image_0000002687958597.png "点击放大") #### 数据一致性校验工具 数据一致性验证工具基于Shell脚本，通过调用Spark-shell启动Scala应用程序。该Scala程序实现了核心的比较逻辑，对比Spark执行查询和通过Spark Native方式执行查询时生成的表数据是否一致。Spark和Spark Native的查询结果。 1. 参考[客户端配置参数]开启Spark Native。 2. 进入"*{客户端安装目录}* /Spark/spark/tool"目录，执行**sh md5compare.sh --help** 可查看使用方法： ``` md5compare.sh --dbname --query-path --spark_tbl_location --native_tbl_location --spark_tbl_name --native_tbl_name --clear_tbl_data --driver_mem --executor_mem --executor_num --executor_core ``` 表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 | 3. 执行示例： ``` 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 ``` ![](https://support.huaweicloud.com/cmpntguide-lts-mrs/public_sys-resources/note_3.0-zh-cn.png) 针对Spark和Native Engine在处理decimal，float以及double时可能存在的小数部分的差异，一致性校验工具统一取到小数点后一位做比较。例如:3.14159，一致性工具在比较时取3.1做比较。 #### 差异说明 开启native engine后，执行结果和spark runtime存在细微差异： 1. 由于spark runtime和native engine在处理decimal的舍入规则不同，造成decimal结果最后一位可能存在差异。 2. 读取json格式的binary类型数据时，由于Spark在写入时进行了base64编码：如 '0101' 写入json文件后为'MDEwMQ=='，Spark runtime读取会base64解码，但native engine读取时未base64解码，需要unbase64()函数解码。 3. Native Engine不支持SET TIME ZONE INTERVAL语法。 #### 常见问题 **问题一：Spark Native引擎运行时使用的内存比Spark高，更容易出现OOM，可考虑开启shuffle prefer spill和降低Batch Size大小。** 1. 服务端配置： 登录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 | 2. 客户端配置： 在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 ``` **decimal溢出检查和精度** - 是否允许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 ```