Fabric Spark 的本机执行引擎

项目
08/08/2024

本机执行引擎是 Microsoft Fabric 中 Apache Spark 作业执行的突破性增强功能。此矢量化引擎通过直接在 Lakehouse 基础结构上运行 Spark 查询来优化这些查询的性能和效率。引擎的无缝集成意味着无需修改代码，并可避免供应商锁定。其支持 Apache Spark API，并且与运行时 1.2 (Spark 3.4) 兼容，适用于 Parquet 和 Delta 格式。无论数据在 OneLake 中的位置如何，或者是否通过快捷方式访问数据，本机执行引擎都会最大限度地提高效率和性能。

本机执行引擎可显著提升查询性能，同时最大限度地降低运营成本。其提供非凡的速度增强，经 TPC-DS 1TB 基准验证，与传统 OSS（开源软件）Spark 相比，性能提高了多达四倍。该引擎擅长管理各种数据处理应用场景，从常规数据引入、批处理作业和 ETL（提取、转换、加载）任务到复杂的数据科学分析和响应式交互查询。用户受益于加速的处理时间、提高的吞吐量和优化的资源利用率。

本机执行引擎基于两个关键 OSS 组件：Velox、Meta 引入的 C++ 数据库加速库和 Apache Gluten (incubating)，这是一个中间层，负责将基于 JVM 的 SQL 引擎的执行卸载到 Intel 引入的本机引擎。

注意

本机执行引擎目前提供公共预览版。有关详细信息，请参阅当前限制。 在预览版的此阶段，不会产生与使用该版本相关的任何额外费用。

何时使用本机执行引擎

本机执行引擎提供了一个解决方案，用于在大型数据集上运行查询；其通过使用底层数据源的本机功能来优化性能，并将通常与传统 Spark 环境中的数据移动和序列化相关的开销降到最低。该引擎支持各种运算符和数据类型，包括汇总哈希聚合、广播嵌套循环联接 (BNLJ) 和精确的时间戳格式。但是，要充分利用引擎的功能，应考虑其最佳用例：

使用 Parquet 和 Delta 格式的数据时，该引擎非常有效，其可以在本机高效地处理这些数据。
涉及复杂转换和聚合的查询可从引擎的列式处理和矢量化功能受益匪浅。
性能增强在查询不通过避免不支持的功能或表达式来触发回退机制的应用场景最为显著。
该引擎非常适合计算密集型查询，而不是简单或 I/O 绑定查询。

有关本机执行引擎支持的运算符和函数的信息，请参阅 Apache Gluten 文档。

启用本机执行引擎

要在预览阶段使用本机执行引擎的完整功能，需要特定的配置。以下过程演示如何为笔记本、Spark 作业定义和整个环境激活此功能。

重要

本机执行引擎当前支持最新的 GA 运行时版本，即 Runtime 1.2（Apache Spark 3.4、Delta Lake 2.4）。

为笔记本或 Spark 作业定义启用

要为单个笔记本或 Spark 作业定义启用本机执行引擎，必须在执行脚本的开头合并必要的配置：

%%configure 
{ 
   "conf": {
       "spark.native.enabled": "true", 
       "spark.shuffle.manager": "org.apache.spark.shuffle.sort.ColumnarShuffleManager" 
   } 
}

对于笔记本，在第一个单元格中插入所需的配置命令。对于 Spark 作业定义，请在 Spark 作业定义的前线包括配置。

本机执行引擎与自定义池集成，这意味着启用此功能会发起一个新会话，通常需要长达两分钟的时间才能启动。

重要

必须在发起 Spark 会话之前完成本机执行引擎的配置。启动 Spark 会话后，spark.shuffle.manager 设置变为不可变，无法进行更改。确保在笔记本的 %%configure 块内或 Spark 会话生成器中为 Spark 作业定义设置这些配置。

在环境级别启用

为了确保统一性能增强，请在与环境关联的所有作业和笔记本上启用本机执行引擎：

导航到环境设置。
转到“Spark 属性”。
填写“Spark 属性”屏幕上的字段，如下图所示。

属性	Value
spark.native.enabled	是
spark.shuffle.manager	org.apache.spark.shuffle.sort.ColumnarShuffleManager

在环境级别启用时，所有后续作业和笔记本都会继承该设置。此继承可确保环境中创建的任何新会话或资源都自动受益于增强的执行功能。

在查询级别控制

在租户、工作区和环境级别上启用本机执行引擎（与 UI 无缝集成）的机制正在积极开发中。此外，还可以针对特定查询禁用本机执行引擎，特别是涉及当前不支持的运算符时（请参阅限制）。要禁用，请将包含该查询的特定单元格的 Spark 配置 spark.gluten.enabled 设置为 false。

%%sql 
SET spark.native.enabled=FALSE;

%%pyspark
spark.conf.set('spark.native.enabled', 'false')

%%spark  
spark.conf.set("spark.native.enabled", "false")

%%sparkr
library(SparkR)
sparkR.conf("spark.native.enabled", "false")

执行禁用本机执行引擎的查询后，必须通过将 spark.gluten.enabled 设置为 true 为后续单元格重新启用它。此步骤是必需的，因为 Spark 按顺序执行代码单元。

%%sql 
SET spark.native.enabled=TRUE;

%%pyspark
spark.conf.set('spark.native.enabled', 'true')

%%spark  
spark.conf.set("spark.native.enabled", "true")

%%sparkr
library(SparkR)
sparkR.conf("spark.native.enabled", "true")

识别引擎执行的操作

有多种方法可确定 Apache Spark 作业中的运算符是否使用本机执行引擎进行处理。

Spark UI 和 Spark 历史记录服务器

访问 Spark UI 或 Spark 历史记录服务器以找到需要检查的查询。在接口中显示的查询计划中，查找以后缀 Transformer、NativeFileScan 或 VeloxColumnarToRowExec 结尾的所有节点名称。该后缀表示本机执行引擎执行了该操作。例如，节点可能标记为 RollUpHashAggregateTransformer、ProjectExecTransformer、BroadcastHashJoinExecTransformer、ShuffledHashJoinExecTransformer 或 BroadcastNestedLoopJoinExecTransformer。

数据帧说明

或者，可以在笔记本中执行 df.explain() 命令以查看执行计划。在输出中，查找相同的 Transformer、NativeFileScan 或 VeloxColumnarToRowExec 后缀。此方法可快速确认是否由本机执行引擎处理特定操作。

回退机制

在某些情况下，本机执行引擎可能由于不支持的功能等原因而无法执行查询。在这些情况下，该操作会回退到传统的 Spark 引擎。此回退机制可确保工作流不会中断。

限制

虽然本机执行引擎增强了 Apache Spark 作业的性能，但请注意其当前限制。

该引擎不支持 Delta 表的分区写入。不支持某些特定于 Delta 的操作，包括合并操作、检查点扫描和删除向量。
某些 Spark 功能和表达式与本机执行引擎不兼容，例如用户定义的函数 (UDF) 和 array_contains 函数，以及 Spark 结构化流式处理。将这些不兼容的操作或函数用作导入库的一部分也会导致 Spark 引擎的回退。
不支持从使用专用终结点的存储解决方案进行扫描。
该引擎不支持 ANSI 模式，因此其搜索并启用 ANSI 模式后，将回退到 vanilla Spark。

在查询中使用日期筛选器时，必须确保比较匹配双方的数据类型，以避免性能问题。不匹配的数据类型可能无法提升查询执行，并可能需要显式强制转换。始终确保比较左侧 (LHS) 和右侧 (RHS) 的数据类型相同，因为不匹配的类型并非总能自动强制转换。如果无法避免类型不匹配，请使用显式强制转换来匹配数据类型，例如 CAST(order_date AS DATE) = '2024-05-20'。本机执行引擎不会加速有不匹配数据类型（需要强制转换）的查询，因此确保类型一致性对于维持性能至关重要。例如，不适用 order_date 为 DATETIME 且字符串为 DATE 的 order_date = '2024-05-20'，而是将 order_date 显式强制转换为 DATE，以确保一致的数据类型并提高性能。

注意

本机执行引擎目前为预览版，你的见解对我们很重要。我们邀请你直接与产品团队分享你的反馈和评估结果。请填写反馈表。我们期待你的宝贵意见，并热切希望详细讨论你的调查结果。

通过