在 Google Kubernetes Engine 上运行 Spark 作业

本文档介绍了如何创建 Managed Service for Apache Spark on GKE 虚拟集群，然后在该集群上运行 Spark 作业。

选项概览

虽然 GKE 上的 Managed Service for Apache Spark 可为容器化环境提供强大的控制功能，但Google Cloud 还提供全代管式和无服务器选项，可简化操作并加快开发速度。如需比较 Spark Managed Service for Apache Spark 部署选项，请参阅确定最佳 Spark 服务。

准备工作

  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.

    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.

    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. 安装 Google Cloud CLI。

  6. 配置 gcloud CLI 以使用您的联合身份。

    如需了解详情，请参阅使用联合身份登录 gcloud CLI。

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

    gcloud init

  8. 您必须已创建标准（非 Autopilot）Google Kubernetes Engine (GKE) 可用区级或区域级集群，并在集群上启用了 Workload Identity。

所需的角色

您需要拥有某些 IAM 角色才能运行本页中的示例。这些角色可能已获授予，具体取决于组织政策。如需查看角色授予情况，请参阅您是否需要授予角色？。

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

用户角色

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

服务账号角色

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

创建虚拟集群

创建 Managed Service for Apache Spark on GKE 虚拟集群作为 Managed Service for Apache Spark 组件的部署平台。它是一种虚拟资源，与 Managed Service for Apache Spark on Compute Engine 集群不同，它不包含单独的 Managed Service for Apache Spark 主节点和工作器虚拟机。

  • 当您创建 Managed Service for Apache Spark on GKE 虚拟集群时，Managed Service for Apache Spark on GKE 会在 GKE 集群中创建节点池。

  • Managed Service for Apache Spark on GKE 作业会在这些节点池上作为 Pod 运行。节点池以及节点池上的 Pod 调度由 GKE 管理。

  • 创建多个虚拟集群。您可以在 GKE 集群上创建和运行多个虚拟集群，以便通过在虚拟集群之间共享节点池来提高资源利用率。

    • 每个虚拟集群：
      • 使用单独的属性（包括 Spark 引擎版本和工作负载身份）创建
      • 在 GKE 集群上的单独 GKE 命名空间中隔离

控制台

Google Cloud 控制台不支持在 GKE 上创建 Managed Service for Apache Spark 集群。

gcloud

设置环境变量，然后在本地或在 Cloud Shell 中运行 gcloud dataproc clusters gke create 命令以创建 Managed Service for Apache Spark on GKE 集群。

  1. 设置环境变量： 

    DP_CLUSTER=Managed Service for Apache Spark on GKE  cluster-name \
  REGION=region \
  GKE_CLUSTER=GKE cluster-name \
  BUCKET=Cloud Storage bucket-name \
  DP_POOLNAME=node pool-name
  PHS_CLUSTER=Managed Service for Apache Spark PHS server name
    注意：

    • DP_CLUSTER：设置 Managed Service for Apache Spark 虚拟集群名称，该名称必须以小写字母开头，后跟最多 54 个小写字母、数字或连字符。不得以连字符结尾。
    • REGION：region 必须与 GKE 集群所在的区域相同。
    • GKE_CLUSTER：现有 GKE 集群的名称。
    • BUCKET：（可选）您可以指定一个 Cloud Storage 存储桶的名称，Managed Service for Apache Spark 将使用该存储分区来分阶段存储制品。如果您未指定存储桶，Managed Service for Apache Spark on GKE 将创建一个暂存存储桶。
    • DP_POOLNAME：要在 GKE 集群中创建的节点池的名称。
    • PHS_CLUSTER：（可选）Managed Service for Apache Spark PHS Server，用于在活跃和已删除的 Managed Service for Apache Spark on GKE 集群上查看 Spark 作业历史记录。PHS 集群必须与 Managed Service for Apache Spark on GKE 虚拟集群位于同一区域。

  2. 运行以下命令： 

    gcloud dataproc clusters gke create ${DP_CLUSTER} \
    --region=${REGION} \
    --gke-cluster=${GKE_CLUSTER} \
    --spark-engine-version=latest \
    --staging-bucket=${BUCKET} \
    --pools="name=${DP_POOLNAME},roles=default" \
    --setup-workload-identity \
    --history-server-cluster=${PHS_CLUSTER}
    注意：

    • --spark-engine-version：Managed Service for Apache Spark 集群上使用的 Spark 映像版本。您可以使用标识符（例如 3、3.1 或 latest），也可以指定完整的次要版本（例如 3.1-dataproc-5）。
    • --staging-bucket：删除此标志，让 Managed Service for Apache Spark on GKE 创建一个暂存存储桶。
    • --pools：此标志用于指定 Managed Service for Apache Spark 将创建或使用的新节点池或现有节点池来执行工作负载。列出 Managed Service for Apache Spark on GKE 节点池设置，以英文逗号分隔，例如：
      --pools=name=dp-default,roles=default,machineType=e2-standard-4,min=0,max=10
      您必须指定节点池 name 和 role。其他节点池设置为可选。您可以使用多个 --pools 标志来指定多个节点池。必须至少一个节点池具有 default 角色。所有节点池都必须位于同一位置。
    • --setup-workload-identity：此标志可启用 Workload Identity 绑定。这些绑定允许 Kubernetes 服务账号 (KSA) 充当虚拟集群的默认 Managed Service for Apache Spark 虚拟机服务账号（数据平面身份）。

REST

在 cluster.create API 请求中完成 virtualClusterConfig。

在使用任何请求数据之前，请先进行以下替换：

  • PROJECT：Google Cloud 项目 ID
  • REGION：Dataproc 虚拟集群区域（与现有 GKE 集群的区域相同）
  • DP_CLUSTER：Dataproc 集群名称
  • GKE_CLUSTER：GKE 集群名称
  • NODE_POOL：节点池名称
  • PHS_CLUSTER：Persistent History Server (PHS) 集群名称
  • BUCKET：（可选）暂存存储桶名称。将此字段留空，让 Managed Service for Apache Spark on GKE 创建一个暂存存储桶。

HTTP 方法和网址：

POST https://dataproc.googleapis.com/v1/projects/project-id/regions/region/clusters

请求 JSON 正文：

{
  "clusterName":"DP_CLUSTER",
  "projectId":"PROJECT",
  "virtualClusterConfig":{
    "auxiliaryServicesConfig":{
      "sparkHistoryServerConfig":{
        "dataprocCluster":"projects/PROJECT/regions/REGION/clusters/PHS_CLUSTER"
      }
    },
    "kubernetesClusterConfig":{
      "gkeClusterConfig":{
        "gkeClusterTarget":"projects/PROJECT/locations/REGION/clusters/GKE_CLUSTER",
        "nodePoolTarget":[
          {
"nodePool":"projects/PROJECT/locations/REGION/clusters/GKE_CLUSTER/nodePools/NODE_POOL",
            "roles":[
              "DEFAULT"
            ]
          }
        ]
      },
      "kubernetesSoftwareConfig":{
        "componentVersion":{
          "SPARK":"latest"
        }
      }
    },
    "stagingBucket":"BUCKET"
  }
}

如需发送您的请求，请展开以下选项之一：

您应该收到类似以下内容的 JSON 响应：

{
  "projectId":"PROJECT",
  "clusterName":"DP_CLUSTER",
  "status":{
    "state":"RUNNING",
    "stateStartTime":"2022-04-01T19:16:39.865716Z"
  },
  "clusterUuid":"98060b77-...",
  "statusHistory":[
    {
      "state":"CREATING",
      "stateStartTime":"2022-04-01T19:14:27.340544Z"
    }
  ],
  "labels":{
    "goog-dataproc-cluster-name":"DP_CLUSTER",
    "goog-dataproc-cluster-uuid":"98060b77-...",
    "goog-dataproc-location":"REGION",
    "goog-dataproc-environment":"prod"
  },
  "virtualClusterConfig":{
    "stagingBucket":"BUCKET",
    "kubernetesClusterConfig":{
      "kubernetesNamespace":"dp-cluster",
      "gkeClusterConfig":{
"gkeClusterTarget":"projects/PROJECT/locations/REGION/clusters/GKE_CLUSTER",
        "nodePoolTarget":[
          {
"nodePool":"projects/PROJECT/locations/REGION/clusters/GKE_CLUSTER/nodePools/NODE_POOL",
            "roles":[
              "DEFAULT"
            ]
          }
        ]
      },
      "kubernetesSoftwareConfig":{
        "componentVersion":{
          "SPARK":"3.1-..."
        },
        "properties":{
          "dpgke:dpgke.unstable.outputOnly.endpoints.sparkHistoryServer":"https://...",
          "spark:spark.eventLog.dir":"gs://BUCKET/.../spark-job-history",
          "spark:spark.eventLog.enabled":"true"
        }
      }
    },
    "auxiliaryServicesConfig":{
      "sparkHistoryServerConfig":{
        "dataprocCluster":"projects/PROJECT/regions/REGION/clusters/PHS_CLUSTER"
      }
    }
  }

提交 Spark 作业

在 Managed Service for Apache Spark on GKE 虚拟集群运行后，请使用 Google Cloud 控制台、gcloud CLI 或 Managed Service for Apache Spark jobs.submit API（通过直接 HTTP 请求或 Cloud 客户端库）提交 Spark 作业。

**gcloud CLI Spark 作业示例：**

gcloud dataproc jobs submit spark \
    --region=${REGION} \
    --cluster=${DP_CLUSTER} \
    --class=org.apache.spark.examples.SparkPi \
    --jars=local:///usr/lib/spark/examples/jars/spark-examples.jar \
    -- 1000

**gcloud CLI PySpark 作业示例：**

gcloud dataproc jobs submit pyspark \
    --region=${REGION} \
    --cluster=${DP_CLUSTER} \
    local:///usr/lib/spark/examples/src/main/python/pi.py \
    -- 10

**gcloud CLI SparkR 作业示例：**

gcloud dataproc jobs submit spark-r \
    --region=${REGION} \
    --cluster=${DP_CLUSTER} \
    local:///usr/lib/spark/examples/src/main/r/dataframe.R

清理