使用 Lightning Engine

Lightning Engine 是 Managed Service for Apache Spark 上 Apache Spark 的高性能执行环境，可加速计算密集型工作负载并降低运行时成本，而无需更改 Spark 代码。它专为运行计算密集型工作负载的数据工程师和分析从业者而设计，这些工作负载使用 Spark DataFrame API、Spark Dataset API 或 Spark SQL 查询。

如果您的工作负载主要受 I/O 限制或严重依赖弹性分布式数据集 (RDD)，则标准引擎可能更合适。

优势

Lightning Engine 的优势包括：

• 加速数据操作：通过优化云端存储空间交互（包括元数据处理、写入工作负载和矢量化 I/O），实现显著的性能提升和成本节省。

• 智能查询执行：利用先进的优化器增强功能，动态减少扫描的数据量、优化数据处理，并生成更高效的执行计划，从而实现更快速、更经济高效的查询。

• 简化的 AI 和机器学习工作负载：缩短基于 GPU 的工作负载的集群启动时间，并使用针对 AI 优化的映像简化在安全环境中的部署。

虽然 Lightning Engine 可显著提升性能，但具体影响因工作负载而异。

与标准引擎的比较

Lightning Engine 是标准引擎的替代方案，用于在 Managed Service for Apache Spark 集群上执行 Spark 作业。下表比较了 Lightning Engine 与标准引擎的激活属性、工作负载适用性和主要优势。

功能	标准引擎	Lightning Engine
CLI 标志	--engine=default 或取消设置该标志	--engine=lightning
适用场景	通用作业、开发和测试	需要大幅加速的企业级工作负载
主要优势	基准性能	优化了云存储交互，实现了智能查询执行

Lightning Engine 要求

Lightning Engine 功能需满足以下要求：

• 映像版本：Lightning Engine 必须与 Managed Service for Apache Spark 映像版本 2.3.3 及更高版本的 2.3 次要映像版本，或 3.0.2 及更高版本的 3.0 次要映像版本（debian13 和 ubuntu24 映像）搭配使用。
• 支持的作业：支持 Spark、PySpark、SparkSQL 和 SparkR。标准引擎在提交到 Lightning Engine 集群的其他服务类型上运行。

原生查询执行

原生查询执行 (NQE) 是 Lightning Engine 的可选组件，可为特定作业提供更深层次的加速。它是一个基于 Apache Gluten 和 Velox 的原生引擎，针对 Google 硬件进行了优化，通过在 JVM 之外运行部分 Spark 查询来提升性能。

建议在以下情况下使用 NQE：

利用 Spark DataFrame API 和 Spark Dataset API 的计算密集型任务，以及从 Parquet、ORC、Apache Iceberg 和 Delta Lake 文件和表中读取数据的 Spark SQL 查询。输出文件格式不会影响其性能。

不建议在以下情况下使用 NQE：

严重依赖弹性分布式数据集 (RDD)、用户定义的函数 (UDF)、大多数 Spark 机器学习 (ML) 库以及因存储访问而导致延迟的 I/O 密集型操作的作业。

ARM 上的原生查询执行

Managed Service for Apache Spark 支持在 ARM 架构上的 Lightning Engine 中执行原生查询，该架构专门针对 C4A ARM 虚拟机上的 Google Axion 处理器进行了优化（请参阅 Managed Service for Apache Spark 支持的 ARM 机器类型），并使用 2.3-ubuntu22-arm 映像。Google Axion C4A 实例是基于 ARM 的虚拟机，构建在 ARM Neoverse V2 核心上，可显著提升性价比。

在 Google Axion ARM 实例上运行 NQE，可将硬件级性价比和高每核内存带宽与软件级原生向量查询加速相结合，从而显著缩短查询运行时长并降低总拥有成本 (TCO)。

ARM 的主要优势包括：

• 速度比开源 Spark 快 2.5 倍到 4 倍：与开源 Apache Spark 相比，分析工作负载的性能提升了 2.5 倍到 4 倍。
• 费用优化：最大限度地发挥 Google Axion C4A 实例的性价比优势（与同类 x86 实例相比，性价比提升高达 65%）。
• 完整的算子和格式对等：与 x86 NQE 算子完全对等，并为 Cloud Storage Parquet、Apache Iceberg 和 Delta Lake 表格式提供原生加速。
• 无需更改代码：使用标准配置标志在 ARM 集群上无缝激活，无需修改 Spark 应用代码。

原生查询执行要求

原生查询执行功能需满足以下要求：

• 执行引擎：NQE 仅适用于在创建集群时启用了 Lightning 引擎的集群。

• 支持的作业：支持 Spark、PySpark、Spark SQL 和 SparkR。对于提交到 Lightning 引擎集群的其他服务类型，标准引擎将运行（不使用 NQE）。

• 无 GPU 和加速器：在 GPU 加速器上提交的启用 NQE 的作业将失败（但可以从不带 NQE 的 Lightning Engine 中受益）。

• 数据类型：不支持以下数据类型的输入：

  • Byte：ORC 和 Parquet
  • 结构体、数组、映射：Parquet

架构和操作系统要求

NQE 要求因处理器架构而异：

架构	支持的机器类型	支持的操作系统和映像
x86	Intel 和 AMD 机器家族	Debian-12 和 Ubuntu-22（映像版本 2.3.3 及更高版本的 2.3 版本）；Debian-13 和 Ubuntu-24（映像版本 3.0.2 及更高版本的 3.0 版本）
ARM	C4A 机器系列（Google Axion ARM 虚拟机）	2.3-ubuntu22-arm 映像（2.3.3 版及更高版本的 2.3 版本）

价格

如需了解价格信息，请参阅 Managed Service for Apache Spark 价格。

创建 Lightning Engine 集群

本部分介绍如何创建 Managed Service for Apache Spark 集群，以便在提交到该集群的 Spark 作业上启用 Lightning Engine。

您还可以在创建集群时在集群上启用原生查询执行 (NQE)，也可以稍后为提交到集群的特定 Spark 作业启用 NQE。

准备工作

1. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Roles required to select or create a project

   • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
   • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.
   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

2. Verify that you have the permissions required to complete this guide.

3. Verify that billing is enabled for your Google Cloud project.

4. Enable the Dataproc API, if it is not already enabled.

   Roles required to enable APIs

   To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

   Enable the API

5. Install the Google Cloud CLI.

   Note: If you installed the gcloud CLI previously, make sure you have the latest version by running gcloud components update.

6. Configure the gcloud CLI to use your federated identity.

   For more information, see Sign in to the gcloud CLI with your federated identity.

7. To initialize the gcloud CLI, run the following command:

   gcloud init

所需的角色

您需要拥有特定的 Identity and Access Management (IAM) 角色，才能创建 Managed Service for Apache Spark 集群并向该集群提交作业。根据组织政策，云项目所有者或服务管理员可能已向您或服务账号授予了这些角色。如需检查角色授予情况，请参阅您是否需要授予角色？

如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

用户角色

如需获得创建 Managed Service for Apache Spark 集群所需的权限，请让管理员向您授予以下 IAM 角色：

• 针对项目的 Dataproc Editor (roles/dataproc.editor)
• Compute Engine 默认服务账号的 Service Account User (roles/iam.serviceAccountUser)

服务账号角色

如需确保 Compute Engine 默认服务账号具有创建 Managed Service for Apache Spark 集群所需的权限，请让您的管理员为 Compute Engine 默认服务账号授予项目的 Dataproc Worker (roles/dataproc.worker) IAM 角色。

重要提示：您必须向 Compute Engine 默认服务账号（而不是向您的用户账号）授予此角色。如果未能向正确的主账号授予此角色，可能会导致权限错误。

创建集群

以下示例展示了如何使用 Google Cloud 控制台、Google Cloud CLI、Dataproc API、Python 版 Cloud 客户端库或 Terraform 创建 Lightning Engine 集群。您还可以使用 Go、Java 和 Node.js Cloud 客户端库创建启用了 Lightning Engine 的集群。

控制台

1. 打开创建集群页面。
2. 点击其他配置以展开该部分。
3. 修改自定义及其他。
4. 在打开的面板中，确认已选中启用 Lightning 引擎复选框。
5. 可选：如需默认为 Spark 作业启用原生执行运行时，请选中启用原生执行复选框。
6. 点击保存。
7. 根据需要配置其他集群设置。
8. 点击创建集群。

创建 ARM 集群：

如需创建具有 Google Axion 和原生查询执行功能的 ARM 集群，请执行以下操作：

1. 在工作器配置中，选择 C4A 机器系列，然后选择 ARM 机器类型。
2. 在映像中，选择 2.3-ubuntu22-arm 映像。
3. 在其他配置 > 自定义和其他中，确认已选择启用 Lightning 引擎和启用原生执行。
4. 点击创建集群。

gcloud

1. 如需创建启用了 Lightning Engine 的集群，请在运行 gcloud dataproc clusters create 命令时使用 --engine=lightning 标志。如需了解详情，请参阅使用 gcloud CLI 创建集群。

   gcloud dataproc clusters create CLUSTER_NAME \
       --region=REGION \
       --engine=lightning \
       --image-version=2.3
   

2. 可选：如需默认为 Spark 作业启用原生执行运行时，请添加 spark:spark.dataproc.lightningEngine.runtime=native 属性。

   gcloud dataproc clusters create CLUSTER_NAME \
       --region=REGION \
       --engine=lightning \
       --image-version=2.3 \
       --properties='spark:spark.dataproc.lightningEngine.runtime=native'
   

3. 创建启用了 NQE 的 ARM 集群：如需创建启用了 Google Axion C4A ARM 节点和 NQE 的 Managed Service for Apache Spark 集群，请运行以下命令：

   gcloud dataproc clusters create CLUSTER_NAME \
       --region=REGION \
       --image-version=2.3-ubuntu22-arm \
       --master-machine-type=c4a-standard-16 \
       --worker-machine-type=c4a-standard-16 \
       --engine=lightning \
       --properties='spark:spark.dataproc.lightningEngine.runtime=native'
   

API

如需创建启用了 Lightning Engine 的集群，请发送 clusters.create 请求。如需了解详情，请参阅使用 REST API 创建集群。

1. 在请求正文中，将 engine 字段设置为 LIGHTNING。

   {
     "projectId": "PROJECT_ID",
     "clusterName": "CLUSTER_NAME",
     "config": {
       "engine": "LIGHTNING",
       "gceClusterConfig": {},
       "softwareConfig": {
         "imageVersion": "2.3"
       }
     }
   }
   

2. 可选：如需默认为所有作业启用原生执行运行时，请添加 spark:spark.dataproc.lightningEngine.runtime 属性。

   {
     "projectId": "PROJECT_ID",
     "clusterName": "CLUSTER_NAME",
     "config": {
       "engine": "LIGHTNING",
       "gceClusterConfig": {},
       "softwareConfig": {
         "imageVersion": "2.3",
         "properties": {
           "spark:spark.dataproc.lightningEngine.runtime": "native"
         }
       }
     }
   }
   

Python

1. 如需创建启用了 Lightning 引擎的集群，请使用 create_cluster 方法，并将集群配置中的 engine 字段设置为 LIGHTNING。如需了解详情，请参阅使用 Python 创建集群。

   from google.cloud import dataproc_v1
   
   def create_lightning_cluster(project_id, region, cluster_name):
       client_options = {"api_endpoint": f"{region}-dataproc.googleapis.com:443"}
       cluster_client = dataproc_v1.ClusterControllerClient(client_options=client_options)
   
       cluster = {
           "project_id": project_id,
           "cluster_name": cluster_name,
           "config": {
               "engine": "LIGHTNING",
               "software_config": {
                   "image_version": "2.3-debian12",
               },
           }
       }
   
       operation = cluster_client.create_cluster(
           project_id=project_id,
           region=region,
           cluster=cluster
       )
       result = operation.result()
       print(f"Cluster created successfully: {result.cluster_name}")
   

2. 可选：如需默认为 Spark 作业启用原生执行运行时，请添加 spark:spark.dataproc.lightningEngine.runtime 属性。

   from google.cloud import dataproc_v1
   
   def create_lightning_native_cluster(project_id, region, cluster_name):
       client_options = {"api_endpoint": f"{region}-dataproc.googleapis.com:443"}
       cluster_client = dataproc_v1.ClusterControllerClient(client_options=client_options)
   
       cluster = {
           "project_id": project_id,
           "cluster_name": cluster_name,
           "config": {
               "engine": "LIGHTNING",
               "software_config": {
                   "image_version": "2.3-debian12",
                   "properties": {
                       "spark:spark.dataproc.lightningEngine.runtime": "native"
                   }
               }
           }
       }
   
       operation = cluster_client.create_cluster(
           project_id=project_id,
           region=region,
           cluster=cluster
       )
       result = operation.result()
       print(f"Cluster created successfully: {result.cluster_name}")
   

Terraform

1. 在 google_dataproc_cluster 资源配置中，将 engine 参数设置为 LIGHTNING。
2. 如需了解详情和高级选项，请参阅 google_dataproc_cluster 资源的官方 Terraform 文档。

验证集群引擎

控制台

1. 在 Google Cloud 控制台中，前往集群详情页面。
2. 验证 Engine 字段中是否列出了 Lightning Engine 值。
3. 如果您启用了原生查询执行，请验证 native 是否列在原生执行字段中。

gcloud

1. 如需验证引擎和 NQE（如果已启用），请运行 gcloud dataproc clusters describe 命令：

   gcloud dataproc clusters describe CLUSTER_NAME --project=PROJECT_ID --region=REGION
   

2. 检查输出中是否包含 engine 和 lightningEngine.runtime 属性：

   clusterName: lightning-engine-cluster
   engine: lightningEngine
   lightningEngine.runtime: native
   

使用 Lightning Engine 提交作业

如果您在创建集群时启用了 Lightning Engine，则在向集群提交 Spark 作业时，系统会默认在作业上启用 Lightning Engine。

为作业启用原生查询执行

如果您在创建 Lightning Engine 集群时启用了原生查询执行 (NQE)，则所有 Spark 作业都会在启用 NQE 的情况下运行，除非您在特定作业上停用 NQE。

如果您在创建 Lightning Engine 集群时未启用 NQE，则可以在提交作业时为作业启用 NQE，如以下示例所示。

gcloud

如需在提交 Spark 作业时启用原生查询执行，请添加 spark.dataproc.lightningEngine.runtime=native 属性：

gcloud dataproc jobs submit spark \
    --cluster=CLUSTER_NAME \
    --region=REGION \
    --properties=spark.dataproc.lightningEngine.runtime=native \
    -- ...

API

如需在提交 Spark 作业时启用原生查询执行，请在请求中添加 spark.dataproc.lightningEngine.runtime 属性：

{
  "job":{
    "placement":{
      "clusterName": ...
    },
    "sparkJob":{
      "mainClass": ...,
      "properties":{
         "spark.dataproc.lightningEngine.runtime":"native"
      }
    }
  }
}

为作业停用原生查询执行

如果您在创建 Lightning Engine 集群时启用了原生查询执行 (NQE)，则所有 Spark 作业都将在启用 NQE 的情况下运行，除非您在特定作业中停用了 NQE。

您可以在提交特定 Spark 作业时停用 NQE，如以下示例所示。

gcloud

如需在提交 Spark 作业到 Lightning Engine 集群时停用原生查询执行，请添加 spark.dataproc.lightningEngine.runtime=default 属性：

gcloud dataproc jobs submit spark \
    --cluster=CLUSTER_NAME \
    --region=REGION \
    --properties=spark.dataproc.lightningEngine.runtime=default \
    -- ...

API

如需在提交 Spark 作业到 Lightning Engine 集群时停用原生查询执行，请添加 spark.dataproc.lightningEngine.runtime=default 属性：

{
  "job":{
    "placement":{
      "clusterName": ...
    },
    "sparkJob":{
      "mainClass": ...,
      "properties":{
         "spark.dataproc.lightningEngine.runtime":"default"
      }
    }
  }
}

验证作业的原生查询执行

将作业提交到 Lightning Engine 集群后，您可以验证是否已为该作业启用原生查询执行。

控制台

1. 在 Google Cloud 控制台中，前往作业页面。
2. 点击任务 ID 以打开作业详情页面。
3. 验证 native 是否列在原生执行字段中。

gcloud

1. 运行 gcloud dataproc jobs describe 命令：

   gcloud dataproc jobs describe JOB_ID --project=PROJECT_ID --region=REGION
   

2. 在属性部分中检查 lightningEngine.runtime 的输出：

   lightningEngine.runtime: native
   

配置参数

下表总结了 Lightning Engine 和原生查询执行的主要配置参数。

参数名称	说明	适用的引擎	默认值	默认值（Lightning 引擎）	用户可覆盖（作业级）	范围
--engine	集群级设置，用于在创建集群期间选择引擎。	集群级	default	lightning	否	集群
spark:spark.dataproc.lightningEngine.runtime	集群级设置，用于在创建集群期间选择 Lightning 引擎运行时。	仅限 Lightning	default	default	否	集群
spark.dataproc.lightningEngine.runtime	在 Lightning Engine 中启用或停用原生查询执行 (NQE)。	仅限 Lightning	default	default	可以。可以设置为 native 或 default。	作业

限制

在以下场景中启用 Native Query Execution 可能会导致异常、Spark 不兼容或工作负载回退到默认 Spark 引擎。

后备值

在以下情况下，原生查询执行可能会导致工作负载回退到 Spark 执行引擎：

• ANSI：如果启用了 ANSI 模式，执行会回退到 Spark。
• 区分大小写模式：原生查询执行仅支持 Spark 默认的不区分大小写模式。如果启用了区分大小写模式，则可能会出现不正确的结果。
• 分区表扫描：仅当路径包含分区信息时，原生查询执行才支持分区表扫描。否则，工作负载会回退到 Spark 执行引擎。

不兼容的行为

在以下情况下，使用原生查询执行可能会导致不兼容的行为或不正确的结果：

• JSON 函数：原生查询执行支持用英文双引号（而非英文单引号）括起来的字符串。使用单引号时会出现不正确的结果。在路径中将 * 与 get_json_object 函数结合使用会返回 NULL。
• Parquet 读取配置：
  • 原生查询执行会将 spark.files.ignoreCorruptFiles 视为设置为默认 false 值，即使设置为 true 也是如此。
  • 原生查询执行会忽略 spark.sql.parquet.datetimeRebaseModeInRead，并且仅返回 Parquet 文件内容。不考虑旧版混合历法与预推格里高利历之间的差异。Spark 结果可能会有所不同。
• NaN：不支持。例如，当您在数值比较中使用 NaN 时，可能会出现意外结果。
• Spark 列式读取：由于 Spark 列式向量与原生查询执行不兼容，可能会发生严重错误。
• 溢出：如果您将 shuffle 分区设置为较大的数字，则溢出到磁盘功能可能会触发 OutOfMemoryException。如果发生这种情况，减少分区数量可以消除此异常。

后续步骤

• 使用 Lightning Engine 加速 Spark 批处理工作负载和会话。
• 查看 Managed Service for Apache Spark 支持的 ARM 机器类型。
• 了解如何调整 Spark 作业以获得最佳性能。