提交 Apache Spark 批处理工作负载

了解如何在 Managed Service for Apache Spark 计算基础架构上提交批量工作负载，该基础架构会根据需要扩缩资源。

准备工作

设置项目，并根据需要授予 Identity and Access Management 角色。

设置项目

根据需要执行以下一个或多个步骤：

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 billing is enabled for your Google Cloud project.

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

根据需要授予 IAM 角色

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

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

用户角色

如需获得提交无服务器批处理工作负载所需的权限，请让您的管理员为您授予以下 IAM 角色：

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

服务账号角色

如需确保 Compute Engine 默认服务账号具有提交无服务器批处理工作负载所需的权限，请让您的管理员向 Compute Engine 默认服务账号授予项目的 Dataproc Worker (roles/dataproc.worker) IAM 角色。

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

提交 Spark 批处理工作负载

您可以使用 Google Cloud 控制台、Google Cloud CLI 或 REST API 创建和提交 Managed Service for Apache Spark 批处理工作负载。

注意：您可以使用控制台中的基于模板创建批处理页面，根据已保存的模板提交批处理作业。如需了解详情，请参阅 Managed Service for Apache Spark 模板。

控制台

1. 在 Google Cloud 控制台中，前往 Managed Service for Apache Spark 作业页面。

   前往 Managed Service for Apache Spark 作业

2. 点击提交作业，打开提交作业页面。

3. 选择并填写以下字段，提交用于计算 pi 近似值的 Spark 批处理工作负载：

   • 为作业选择计算类型：选择无服务器 Spark。
   • 无服务器作业详情：
     • 任务 ID：为批处理工作负载指定 ID（或接受默认 ID）。此值必须为 4 到 63 个小写字符，以小写字母或数字开头和结尾，并且只能包含小写字母、数字和连字符 (/[a-z0-9-]/)。
     • 运行时配置文件：在区域列表中选择一个区域，然后在运行时配置文件列表中选择默认服务配置（或已保存的运行时配置文件）。
     • 配置运行时配置文件（可选）：在摘要卡片中，点击配置（或点击运行时配置文件菜单中的创建新配置文件），打开运行时配置文件配置面板，您可以在其中自定义以下设置，然后点击使用配置（或保存新配置文件）：
       • 层级和执行器配置：选择计算层级（高级或标准）和执行器形状。
       • 其他配置：
         • 运行时配置：点击修改以确认或选择 3.0 Spark 运行时版本，或指定自定义 Spark 映像、Cloud Storage 暂存存储桶或 Python 软件包仓库。
         • Metastore 配置：点击修改，将工作负载配置为使用 Lakehouse 目录或外部自行管理的 Dataproc Metastore。
         • 网络和安全：点击修改以选择服务账号（默认情况下，工作负载使用 Compute Engine 默认服务账号运行，或者您可以指定具有 Dataproc Worker 角色的自定义服务账号），并在工作负载区域中选择子网（Managed Service for Apache Spark 会在指定子网上启用专用 Google 访问通道 (PGA)；请参阅 Managed Service for Apache Spark 网络配置）。
   • 文件和制品：
     • 服务类型：选择 Spark (Scala/Java)。
     • 驱动程序配置：选择主类。
     • 主类：

       org.apache.spark.examples.SparkPi

     • 参数：输入 1000，然后按 Enter。
     • Jar 文件（此文件预安装在 Managed Service for Apache Spark 执行环境中）：

       file:///usr/lib/spark/examples/jars/spark-examples.jar

   • 作业执行与属性：
     • Spark 属性：点击添加属性，输入要在 Spark 批处理工作负载上设置的受支持的 Spark 属性的键（属性名称）和值。注意：与 Managed Service for Apache Spark 集群属性不同，Managed Service for Apache Spark 工作负载属性不包含 spark: 前缀。
     • 标签：可选：点击添加标签，将键和值标签对附加到工作负载。

4. 点击提交以运行 Spark 批处理工作负载。

gcloud

如需提交 Spark 批处理工作负载以计算 pi 的近似值，请在终端窗口或 Cloud Shell 中本地运行以下 gcloud CLI gcloud dataproc batches submit spark 命令。

gcloud dataproc batches submit spark \
    --region=REGION \
    --version=3.0 \
    --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \
    --class=org.apache.spark.examples.SparkPi \
    -- 1000

请替换以下内容：

• REGION: 指定将在其中运行工作负载的区域。
• 其他选项：您可以添加 gcloud dataproc batches submit spark 标志来指定其他工作负载选项和 Spark 属性。
  • --jars：示例 JAR 文件预安装在 Spark 执行环境中，传递给 SparkPi 工作负载的 1000 命令参数指定了 pi 估算逻辑的 1000 次迭代（工作负载输入参数包含在“--”后面）。
  • --subnet：您可以添加此标志，以指定会话区域中子网的名称。如果您未指定子网，Managed Service for Apache Spark 会选择会话区域中的 default 子网。Managed Service for Apache Spark 会在子网中启用专用 Google 访问通道 (PGA)。如需了解网络连接要求，请参阅 Managed Service for Apache Spark 网络配置。
  • --tags：您可以添加此标志来指定用于流量控制的网络标记。使用网络标记来限制连接。在生产环境中，建议的做法是将防火墙规则限制为 Spark 工作负载使用的 IP 地址。
  • --properties：您可以添加此标志，输入受支持的 Spark 属性，供 Spark 批处理工作负载使用。
  • --deps-bucket：您可以添加此标志以指定一个 Cloud Storage 存储桶，Managed Service for Apache Spark 将在其中上传工作负载依赖项。存储桶的 gs:// URI 前缀不是必需的；您可以指定存储桶路径或存储桶名称。在运行批量工作负载之前，Managed Service for Apache Spark 会将本地文件上传到存储桶中的 /dependencies 文件夹。注意：如果您的批处理工作负载引用本地机器上的参考文件，则此标志是必需的。
  • --ttl：您可以添加 --ttl 标志来指定批处理生命周期的时长。当工作负载超过此时长时，系统会无条件终止该工作负载，而不会等待正在进行的工作完成。使用 s、m、h 或 d（秒、分钟、小时或天）后缀指定时长。最小值为 10 分钟 (10m)，最大值为 14 天 (14d)。
    • 1.1 或 2.0 运行时批处理：如果未为 1.1 或 2.0 运行时批处理工作负载指定 --ttl，则允许该工作负载运行，直到其自然退出（如果未退出，则永久运行）。
    • 2.1+ 运行时批处理：如果未为 2.1 或更高版本的运行时批处理工作负载指定 --ttl，则默认为 4h。
  • --service-account：您可以指定用于运行工作负载的服务账号。如果您未指定服务账号，则工作负载将在 Compute Engine 默认服务账号下运行。 您的服务账号必须具有 Dataproc Worker 角色。
  • Hive Metastore：以下命令将批处理工作负载配置为使用具有标准 Spark 配置的自行管理式 Hive Metastore。

    gcloud dataproc batches submit spark\
        --properties=spark.sql.catalogImplementation=hive,spark.hive.metastore.uris=METASTORE_URI,spark.hive.metastore.warehouse.dir=WAREHOUSE_DIR> \
        other args ...
            

  • Persistent History Server：
    1. 以下命令会在单节点 Managed Service for Apache Spark 集群上创建 PHS。PHS 必须位于您运行批处理工作负载的区域，并且 Cloud Storage bucket-name 必须存在。

       gcloud dataproc clusters create PHS_CLUSTER_NAME \
           --region=REGION \
           --single-node \
           --enable-component-gateway \
           --properties=spark:spark.history.fs.logDirectory=gs://bucket-name/phs/*/spark-job-history
                    

    2. 提交批处理工作负载，并指定正在运行的 Persistent History Server。

       gcloud dataproc batches submit spark \
           --region=REGION \
           --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \
           --class=org.apache.spark.examples.SparkPi \
           --history-server-cluster=projects/project-id/regions/region/clusters/PHS-cluster-name \
           -- 1000
                     

  • 运行时版本：使用 --version 标志为工作负载指定 Managed Service for Apache Spark 运行时版本。

    gcloud dataproc batches submit spark \
        --region=REGION \
        --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \
        --class=org.apache.spark.examples.SparkPi \
        --version=VERSION
        -- 1000
                

API

本部分介绍如何使用 Managed Service for Apache Spark batches.create` 创建批处理工作负载以计算 pi 的近似值

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

• project-id： Google Cloud 项目 ID。
• region：Managed Service for Apache Spark 将在其中运行工作负载的 Compute Engine 区域。

注意：

• PROJECT_ID：您的 Google Cloud 项目 ID。 项目 ID 列在 Google Cloud 控制台信息中心的项目信息部分中。
• REGION：会话区域。

HTTP 方法和网址：

POST https://dataproc.googleapis.com/v1/projects/project-id/locations/region/batches

请求 JSON 正文：

{
  "sparkBatch":{
    "args":[
      "1000"
    ],
    "runtimeConfig": {
      "version": "2.3",
    },
    "jarFileUris":[
      "file:///usr/lib/spark/examples/jars/spark-examples.jar"
    ],
    "mainClass":"org.apache.spark.examples.SparkPi"
  }
}

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

curl（Linux、macOS 或 Cloud Shell）

注意：以下命令假定您已使用您的用户账号通过运行 gcloud init 或 gcloud auth login 登录 gcloud CLI，或者使用了 Cloud Shell，这会使您自动登录 gcloud CLI。您可以运行 gcloud auth list 来检查当前活跃的账号。

将请求正文保存在名为 request.json 的文件中，然后执行以下命令：

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://dataproc.googleapis.com/v1/projects/project-id/locations/region/batches"

PowerShell (Windows)

注意：以下命令假定您已使用您的用户账号通过运行 gcloud init 或 gcloud auth login 登录 gcloud CLI。您可以运行 gcloud auth list 来检查当前活跃的账号。

将请求正文保存在名为 request.json 的文件中，然后执行以下命令：

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://dataproc.googleapis.com/v1/projects/project-id/locations/region/batches" | Select-Object -Expand Content

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

{
"name":"projects/project-id/locations/region/batches/batch-id",
  "uuid":",uuid",
  "createTime":"2021-07-22T17:03:46.393957Z",
  "sparkBatch":{
    "mainClass":"org.apache.spark.examples.SparkPi",
    "args":[
      "1000"
    ],
    "jarFileUris":[
      "file:///usr/lib/spark/examples/jars/spark-examples.jar"
    ]
  },
  "runtimeInfo":{
    "outputUri":"gs://dataproc-.../driveroutput"
  },
  "state":"SUCCEEDED",
  "stateTime":"2021-07-22T17:06:30.301789Z",
  "creator":"account-email-address",
  "runtimeConfig":{
    "version":"2.3",
    "properties":{
      "spark:spark.executor.instances":"2",
      "spark:spark.driver.cores":"2",
      "spark:spark.executor.cores":"2",
      "spark:spark.app.name":"projects/project-id/locations/region/batches/batch-id"
    }
  },
  "environmentConfig":{
    "peripheralsConfig":{
      "sparkHistoryServerConfig":{
      }
    }
  },
  "operation":"projects/project-id/regions/region/operation-id"
}

工作负载如何计算 Pi：Spark 工作负载使用 Monte Carlo 方法估算 Pi 的值。 它在坐标平面上生成 `x,y` 点，模拟由一个单位正方形包围的圆。输入参数 (`1000`) 决定要生成的 x,y 对的数量；生成的数量越多，估算的准确性就越高。如需了解详情，请参阅使用 Monte Carlo 方法估算 Pi 并参阅 GitHub 上的 JavaSparkPi.java。

估算工作负载费用

Managed Service for Apache Spark 工作负载会消耗数据计算单元 (DCU) 和 shuffle 存储资源。如需查看输出 Managed Service for Apache Spark UsageMetrics 以估算工作负载资源消耗和费用的示例，请参阅 Managed Service for Apache Spark 价格。

后续步骤

了解：

• 使用 Google Cloud Data Agent Kit 在 VS Code 中管理数据工作负载生命周期
• 使用 Data Agent Kit 构建数据流水线
• 在 BigQuery Studio 笔记本中运行 PySpark 代码
• Managed Service for Apache Spark 配额
• Spark 属性