使用 Lightning Engine

Lightning Engine 是新一代 Apache Spark 性能，引入了独有的增强功能，旨在大幅提升性能、成本效益和运营稳定性。

福利

Lightning Engine 的优势包括：

  • 加快数据操作速度：通过优化云存储互动（包括 元数据处理、写入工作负载和向量化 I/O），显著提升性能并 节省成本。

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

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

虽然 Lightning Engine 可大幅提升性能，但具体影响因工作负载而异。它最适合利用 Spark Dataframe API、Spark Dataset API 和 Spark SQL 查询的计算密集型任务，而不是 I/O 绑定操作。

与标准引擎的比较

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

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

要求

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

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

原生查询执行

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

建议将 NQE 用于：
利用 Spark Dataframe API 和 Spark Dataset API 的计算密集型任务，以及从 Parquet 和 ORC 文件读取数据的 Spark SQL 查询。输出文件格式不会影响其性能。
不建议将 NQE 用于：
严重依赖弹性分布式数据集 (RDD)、用户定义函数 (UDF)、大多数 Spark 机器学习 (ML) 库以及因存储访问而延迟的 I/O 绑定操作的作业。

要求

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

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

  • 操作系统：仅支持 Debian-12 和 Ubuntu-22 操作系统 。使用任何其他操作系统的启用 NQE 的作业都会失败。

  • 支持的作业：支持 Spark、PySpark、SparkSQL 和 SparkR。标准引擎将在提交到 Lightning Engine 集群的其他作业类型上运行（不使用 NQE）。

  • 机器类型：仅支持使用 Intel 或 AMD 处理器的机器家族。 使用 ARM 处理器的启用 NQE 的作业将会失败（但可以从不使用 NQE 的 Lightning Engine 中受益）。

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

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

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

价格

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

创建 Lightning Engine 集群

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

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

准备工作

  1. 登录您的 Google Cloud 账号。如果您是新手 Google Cloud， 请创建一个账号来评估我们的产品在 实际场景中的表现。新客户还可获享 $300 赠金，用于 运行、测试和部署工作负载。

  2. 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.

    Go to project selector

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

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

  5. Enable the Dataproc API.

    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

  6. 安装 Google Cloud CLI。

  7. 如果您使用的是外部身份提供方 (IdP)，则必须先使用联合身份登录 gcloud CLI。

  8. 如需初始化 gcloud CLI，请运行以下命令： 

    gcloud init

  9. 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.

    Go to project selector

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

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

  12. Enable the Dataproc API.

    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

  13. 安装 Google Cloud CLI。

  14. 如果您使用的是外部身份提供方 (IdP)，则必须先使用联合身份登录 gcloud CLI。

  15. 如需初始化 gcloud CLI，请运行以下命令： 

    gcloud init

所需的角色

创建 Managed Service for Apache Spark 集群并向该集群提交作业需要某些 IAM 角色。根据组织政策，云项目所有者或服务管理员可能已向您或服务帐号授予这些角色。如需查看 角色授予情况，请参阅您是否需要授予 角色？。

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

用户角色

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

服务账号角色

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

创建集群

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

控制台

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

gcloud CLI

  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'

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 Engine 的集群，请使用 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. 如需了解详情和高级选项，请参阅资源的官方 Terraform 文档。google_dataproc_cluster

验证集群引擎

控制台

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

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

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

API

如需在向 Lightning Engine 集群提交Spark 作业时停用原生查询执行，请添加 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 Execution 字段中是否列出了 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) 用户可替换（作业级） 范围
--engine 集群级设置，用于在创建集群期间选择引擎。 集群范围 default lightning 否 集群
spark:spark.dataproc.lightningEngine.runtime 集群级设置，用于在创建集群期间选择 Lightning 引擎运行时。 仅限 Lightning default default 否 集群
spark.dataproc.lightningEngine.runtime 在 Lightning Engine 中启用或停用原生查询执行 (NQE)。 仅限 Lightning default default 可以。可以设置为 native 或 default。 作业

限制

在以下情况下启用原生查询执行可能会导致异常、Spark 不兼容或工作负载回退到默认 Spark 引擎。

后备广告

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

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

不兼容的行为

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

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

后续步骤