运行 Apache Spark 批处理工作负载

了解如何使用 Dataproc Serverless 在 Dataproc 管理的计算基础架构（可按需扩缩资源）上提交批量工作负载。

准备工作

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

  2. 在 Google Cloud Console 中的项目选择器页面上，选择或创建一个 Google Cloud 项目。

    转到“项目选择器”

  3. 确保您的 Google Cloud 项目已启用结算功能。

  4. 启用 Dataproc API。

    启用 API

    5.

  5. 在 Google Cloud Console 中的项目选择器页面上，选择或创建一个 Google Cloud 项目。

    转到“项目选择器”

  6. 确保您的 Google Cloud 项目已启用结算功能。

  7. 启用 Dataproc API。

    启用 API

    8.

提交 Spark 批处理工作负载

控制台

  1. 在 Google Cloud 控制台中，前往 Dataproc 批次。点击创建以打开创建批次页面。

  2. 选择并填写页面上的以下字段，以提交用于计算 pi 的近似值的 Spark 批量工作负载：

    • 批次信息：
      • 批处理 ID：指定批处理工作负载的 ID。此值必须为 4-63 个小写字符。有效字符为 /[a-z][0-9]-/。
      • 区域：选择将在其中运行工作负载的区域。
    • 容器：
      • 批处理类型：Spark。
      • 运行时版本：选择默认运行时版本。您可以选择指定非默认 Dataproc Serverless 运行时版本。
      • 主类： 
        org.apache.spark.examples.SparkPi
      • Jar 文件（此文件已预安装在 Dataproc Serverless Spark 执行环境中）。 
        file:///usr/lib/spark/examples/jars/spark-examples.jar
      • 参数：1000。
    • 执行配置：您可以指定用于运行工作负载的服务帐号。如果您未指定服务帐号，则系统会在 Compute Engine 默认服务帐号下运行工作负载。
    • 网络配置：必须为专用 Google 访问通道启用 Dataproc Serverless for Spark 工作负载的 VPC 子网，并满足 Dataproc Serverless for Spark 网络配置中列出的其他要求。子网列表会显示所选网络中为专用 Google 访问通道启用的子网。
    • 属性：输入要在 Spark 批量工作负载上设置的受支持的 Spark 属性的 Key（属性名称）和 Value。注意：与 Compute Engine 集群属性上的 Dataproc 不同，适用于 Spark 工作负载属性的 Dataproc Serverless 不包含 spark: 前缀。
    • 其他选项：

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

gcloud

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

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

注意：

  • REGION: 指定将运行工作负载的区域。
  • 子网：必须为专用 Google 访问通道启用 Dataproc Serverless for Spark 工作负载的 VPC 子网，并满足 Dataproc Serverless for Spark 网络配置中列出的其他要求。如果没有为专用 Google 访问通道启用 gcloud dataproc batches submit 命令中所指定区域的 default 网络子网，您必须执行以下操作之一：
    • 为区域的默认网络子网启用专用 Google 访问通道，或
    • 在命令中使用 --subnet=[SUBNET_URI] 标志来指定已启用专用 Google 访问通道的子网。您可以运行 gcloud compute networks describe [NETWORK_NAME] 命令以列出网络中子网的 URI。
  • --jars：示例 JAR 文件已预安装在 Spark 执行环境中，传递给 SparkPi 工作负载的 1000 命令参数指定 pi 估算逻辑的 1,000 次迭代（工作负载输入参数包含在“--”后面）。
  • --properties：您可以添加 --properties 标志，以输入您希望 Spark 批量工作负载使用的受支持的 Spark 属性。
  • --deps-bucket：您可以添加此标志，以指定 Dataproc Serverless 将在其中上传工作负载依赖项的 Cloud Storage 存储桶。存储桶的 gs:// URI 前缀不是必需的；您可以指定存储桶路径或存储桶名称，例如“mybucketname”。在运行批量工作负载之前，Dataproc Serverless for 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。
  • 其他选项：您可以添加 gcloud dataproc batches submit 命令标志来指定其他工作负载选项和 Spark 属性。
    • Hive Metastore：以下命令使用标准 Spark 配置将批量工作负载配置为使用外部自行管理的 Hive Metastore。 
      gcloud dataproc batches submit \
    --properties=spark.sql.catalogImplementation=hive,spark.hive.metastore.uris=METASTORE_URI,spark.hive.metastore.warehouse.dir=WAREHOUSE_DIR> \
    other args ...
    • 永久性历史记录服务器：
      1. 以下命令会在单节点 Dataproc 集群上创建 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. 提交批量工作负载，指定正在运行的永久性历史记录服务器。
        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 标志为工作负载指定 Dataproc 无服务器运行时版本。
      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

本部分介绍如何使用 Dataproc Serverless for Spark batches.create 创建批量工作负载来计算 pi 的近似值

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

  • project-id：Google Cloud 项目 ID。
  • region：Dataproc Serverless 将在其中运行工作负载的 Compute Engine 区域。
    • 注意：

    HTTP 方法和网址： 

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

    请求 JSON 正文：

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

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

    您应该收到类似以下内容的 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":{
    "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"
}

估算工作负载费用

Dataproc Serverless for Spark 工作负载会使用数据计算单元 (DCU) 和 shuffle 存储资源。如需查看输出 Dataproc UsageMetrics 以估算工作负载资源用量和费用的示例，请参阅 Dataproc 无服务器价格。

后续步骤

了解：