运行 Apache Spark 批处理工作负载

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

准备工作

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

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

   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

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

4. 启用 Dataproc API。

   启用 API

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

   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

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

7. 启用 Dataproc API。

   启用 API

提交 Spark 批处理工作负载

控制台

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

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

   • 批次信息：
     • 批处理 ID：指定批处理工作负载的 ID。此值必须为 4-63 个小写字符。有效字符为 /[a-z][0-9]-/。
     • 区域：选择将在其中运行工作负载的区域。
   • 容器：
     • 批次类型：Spark。
     • 运行时版本：选择默认运行时版本。您可以选择指定非默认的 Dataproc 无服务器运行时版本。
     • 主类：

       org.apache.spark.examples.SparkPi

     • Jar 文件（此文件已预安装在 Dataproc Serverless Spark 执行环境中）。

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

     • 参数：1,000。
   • 执行配置：您可以指定用于运行工作负载的服务帐号。如果您未指定服务帐号，工作负载将在 Compute Engine 默认服务帐号下运行。
   • 网络配置：必须为执行 Dataproc Serverless for Spark 工作负载的 VPC 子网启用专用 Google 访问通道，并满足 Dataproc Serverless for Spark 网络配置中列出的其他要求。子网列表会显示所选网络中已启用专用 Google 访问通道的子网。
   • 属性：输入要在 Spark 批处理工作负载上设置的受支持的 Spark 属性的 Key（属性名称）和 Value。注意：与 Compute Engine 上的 Dataproc 集群属性不同，Dataproc Serverless for Spark 工作负载属性不包含 spark: 前缀。
   • 其他选项：
     • 您可以将批量工作负载配置为使用自行管理的外部 Hive Metastore 。
     • 您可以使用永久性历史记录服务器 (PHS)。PHS 必须位于运行批量工作负载的区域。

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: 指定将运行工作负载的区域。
• 子网：必须为执行 Dataproc Serverless for Spark 工作负载的 VPC 子网启用专用 Google 访问通道，并满足 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 估算逻辑的 1000 次迭代（工作负载输入参数包含在“--”后面）。
• --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 区域。
注意：
• Custom-container-image:：使用 Docker 映像命名格式指定自定义容器映像：{hostname}/{project-id}/{image}:{tag}，例如“gcr.io/my-project-id/my-image:1.0.1”。 注意：您必须在 Container Registry 上托管自定义容器。
• 子网：必须为执行 Dataproc Serverless for Spark 工作负载的 VPC 子网启用专用 Google 访问通道 ，并满足 Dataproc Serverless for Spark 网络配置中列出的其他要求。如果没有为专用 Google 访问区域启用 default 网络的子网，您必须执行以下操作之一：
  1. 为区域的默认网络子网启用专用 Google 访问通道，或
  2. 使用 ExecutionConfig.subnetworkUri 字段指定启用了专用 Google 访问通道的子网。您可以运行 gcloud compute networks describe [NETWORK_NAME] 命令以列出网络中子网的 URI。
• sparkBatch.jarFileUris：示例 jar 文件预安装在 Spark 执行环境中。“1000”sparkBatch.args 会传递给 SparkPi 工作负载，并指定 pi 估算逻辑的 1000 次迭代。
• Spark properties：您可以使用 RuntimeConfig.properties 字段输入您希望 Spark 批量工作负载使用的受支持的 Spark 属性。
• --ttl：您可以使用 EnvironmentConfig.ttl 字段指定批次生命周期的时长。当工作负载超过此时长时，将无条件终止，而不会等待正在进行的工作完成。将时长指定为 Duration 的 JSON 表示法。 最小值为 10 分钟，最大值为 14 天。
  • 1.1 或 2.0 运行时批次：如果没有为 1.1 或 2.0 运行时批量工作负载指定 --ttl，则允许工作负载一直运行，直到其自然退出（如果未退出，则一直运行）。
  • 2.1+ 运行时批量：如果没有为 2.1 或更高版本的运行时批量工作负载指定 --ttl，则默认为 4 小时。
• 其他选项：
  • 配置批处理工作负载以使用外部自行管理式 Hive Metastore。
  • 使用永久性历史记录服务器 (PHS)。PHS 必须位于运行批量工作负载的区域。
  • 在 batches.create 请求中使用 RuntimeConfig.version 字段，以指定非默认的 Dataproc 无服务器运行时版本

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"
  }
}

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

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":{
    "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。

估算工作负载费用

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

后续步骤

了解：

• Dataproc 无服务器配额
• Spark 属性