Spark 独立模式

• 安全
• 在集群上安装 Spark 独立模式
• 手动启动集群
• 集群启动脚本
• 资源分配和配置概述
• 将应用程序连接到集群
• 客户端属性
• 启动 Spark 应用程序
  • Spark 协议
  • REST API
• 资源调度
• Executor 调度
• Stage 级别调度概述
  • 注意事项
• 监控和日志记录
• 与 Hadoop 并行运行
• 为网络安全配置端口
• 高可用性
  • 使用 ZooKeeper 的备用 Master
  • 使用本地文件系统的单节点恢复

除了在 Mesos 或 YARN 集群管理器上运行外，Spark 还提供了一种简单的独立部署模式。 您可以手动启动主节点和工作节点来启动独立集群，也可以使用我们提供的 启动脚本。 也可以在一台机器上运行这些守护程序以进行测试。

安全

身份验证等安全功能默认情况下未启用。 部署开放给互联网或不受信任网络的集群时，务必保护对集群的访问，以防止未经授权的应用程序在集群上运行。 在运行 Spark 之前，请参阅Spark 安全和本文档中的特定安全部分。

在集群上安装 Spark 独立模式

要安装 Spark 独立模式，只需将 Spark 的编译版本放在集群上的每个节点上即可。 您可以通过每个版本获取 Spark 的预构建版本，也可以自行构建。

手动启动集群

您可以通过执行以下命令来启动独立的 master 服务器

./sbin/start-master.sh

启动后，master 将打印出自身的 spark://HOST:PORT URL，您可以使用它将工作节点连接到 master，或者将其作为“master”参数传递给 SparkContext。 您还可以在 master 的 Web UI 上找到此 URL，默认情况下为 http://localhost:8080。

同样，您可以启动一个或多个工作节点并通过以下命令将它们连接到 master

./sbin/start-worker.sh <master-spark-URL>

启动工作节点后，查看 master 的 Web UI（默认情况下为 http://localhost:8080）。 您应该在那里看到新节点，以及它的 CPU 数量和内存（减去为操作系统保留的 1 GB）。

最后，可以将以下配置选项传递给 master 和 worker

集群启动脚本

要使用启动脚本启动 Spark 独立集群，您应该在 Spark 目录中创建一个名为 conf/workers 的文件，该文件必须包含您要启动 Spark worker 的所有机器的主机名，每行一个。 如果 conf/workers 不存在，则启动脚本默认为单台机器 (localhost)，这对于测试很有用。 请注意，master 机器通过 ssh 访问每个 worker 机器。 默认情况下，ssh 并行运行，需要设置无密码（使用私钥）访问。 如果您没有无密码设置，您可以设置环境变量 SPARK_SSH_FOREGROUND 并为每个 worker 串行提供密码。

设置好此文件后，您可以使用以下 shell 脚本启动或停止集群，这些脚本基于 Hadoop 的部署脚本，位于 SPARK_HOME/sbin 中

• sbin/start-master.sh - 在运行脚本的机器上启动 master 实例。
• sbin/start-workers.sh - 在 conf/workers 文件中指定的每台机器上启动一个 worker 实例。
• sbin/start-worker.sh - 在运行脚本的机器上启动 worker 实例。
• sbin/start-connect-server.sh - 在运行脚本的机器上启动 Spark Connect server。
• sbin/start-all.sh - 启动 master 和多个 worker，如上所述。
• sbin/stop-master.sh - 停止通过 sbin/start-master.sh 脚本启动的 master。
• sbin/stop-worker.sh - 停止运行脚本的机器上的所有 worker 实例。
• sbin/stop-workers.sh - 停止 conf/workers 文件中指定的机器上的所有 worker 实例。
• sbin/stop-connect-server.sh - 停止运行脚本的机器上的所有 Spark Connect server 实例。
• sbin/stop-all.sh - 停止 master 和 worker，如上所述。

请注意，这些脚本必须在您要运行 Spark master 的机器上执行，而不是在您的本地机器上执行。

您可以通过在 conf/spark-env.sh 中设置环境变量来选择性地进一步配置集群。 通过从 conf/spark-env.sh.template 开始创建此文件，然后将其复制到所有 worker 机器，以使设置生效。 以下设置可用

注意： 启动脚本目前不支持 Windows。 要在 Windows 上运行 Spark 集群，请手动启动 master 和 worker。

SPARK_MASTER_OPTS 支持以下系统属性

SPARK_WORKER_OPTS 支持以下系统属性

资源分配和配置概述

请务必阅读配置页面上的“自定义资源调度和配置概述”部分。本节仅讨论资源调度的 Spark Standalone 特定方面。

Spark Standalone 有两个部分，第一部分是配置 Worker 的资源，第二部分是为特定应用程序分配资源。

用户必须配置 Worker 以拥有一组可用资源，以便它可以将它们分配给 Executor。 spark.worker.resource.{resourceName}.amount 用于控制 Worker 分配的每个资源的数量。用户还必须指定 spark.worker.resourcesFile 或 spark.worker.resource.{resourceName}.discoveryScript，以指定 Worker 如何发现其分配的资源。请参阅上面每个资源的描述，以了解哪种方法最适合您的设置。

第二部分是在 Spark Standalone 上运行应用程序。标准 Spark 资源配置的唯一特殊情况是您在客户端模式下运行 Driver。对于客户端模式下的 Driver，用户可以通过 spark.driver.resourcesFile 或 spark.driver.resource.{resourceName}.discoveryScript 指定其使用的资源。如果 Driver 与其他 Driver 在同一主机上运行，请确保资源文件或发现脚本仅返回与其他在同一节点上运行的 Driver 不冲突的资源。

请注意，用户在提交应用程序时无需指定发现脚本，因为 Worker 将使用分配给它的资源启动每个 Executor。

将应用程序连接到集群

要在 Spark 集群上运行应用程序，只需将 Master 的 spark://IP:PORT URL 作为参数传递给 SparkContext 构造函数。

要针对集群运行交互式 Spark Shell，请运行以下命令

./bin/spark-shell --master spark://IP:PORT

您还可以传递选项 --total-executor-cores <numCores> 来控制 spark-shell 在集群上使用的内核数量。

客户端属性

Spark 应用程序支持以下特定于独立模式的配置属性

启动 Spark 应用程序

Spark 协议

spark-submit 脚本提供了一种最直接的方式将编译后的 Spark 应用程序提交到集群。对于独立集群，Spark 目前支持两种部署模式。在 client 模式下，Driver 与提交应用程序的客户端在同一进程中启动。然而，在 cluster 模式下，Driver 从集群内部的某个 Worker 进程启动，并且客户端进程在完成提交应用程序的职责后立即退出，而无需等待应用程序完成。

如果您的应用程序是通过 Spark submit 启动的，那么应用程序 jar 会自动分发到所有 worker 节点。 对于您的应用程序依赖的任何其他 jar，您应该通过 --jars 标志使用逗号作为分隔符指定它们（例如 --jars jar1,jar2）。 要控制应用程序的配置或执行环境，请参阅 Spark 配置。

此外，standalone cluster 模式支持在应用程序以非零退出代码退出时自动重新启动您的应用程序。 要使用此功能，您可以在启动应用程序时将 --supervise 标志传递给 spark-submit。 然后，如果您想杀死一个重复失败的应用程序，您可以通过以下方式进行操作

./bin/spark-class org.apache.spark.deploy.Client kill <master url> <driver ID>

您可以在独立 Master Web UI（位于 http://<master url>:8080）中找到 Driver ID。

REST API

如果启用了 spark.master.rest.enabled，Spark Master 会通过 http://[host:port]/[version]/submissions/[action] 提供额外的 REST API，其中 host 是 Master 主机，port 是 spark.master.rest.port 指定的端口号（默认值：6066），version 是协议版本，当前为 v1，action 是以下支持的操作之一。

以下是使用 pi.py 和 REST API 的 curl CLI 命令示例。

$ curl -XPOST http://IP:PORT/v1/submissions/create \
--header "Content-Type:application/json;charset=UTF-8" \
--data '{
  "appResource": "",
  "sparkProperties": {
    "spark.master": "spark://master:7077",
    "spark.app.name": "Spark Pi",
    "spark.driver.memory": "1g",
    "spark.driver.cores": "1",
    "spark.jars": ""
  },
  "clientSparkVersion": "",
  "mainClass": "org.apache.spark.deploy.SparkSubmit",
  "environmentVariables": { },
  "action": "CreateSubmissionRequest",
  "appArgs": [ "/opt/spark/examples/src/main/python/pi.py", "10" ]
}'

以下是上述 create 请求的 REST API 的响应。

{
  "action" : "CreateSubmissionResponse",
  "message" : "Driver successfully submitted as driver-20231124153531-0000",
  "serverSparkVersion" : "3.5.1",
  "submissionId" : "driver-20231124153531-0000",
  "success" : true
}

资源调度

独立集群模式目前仅支持跨应用程序的简单 FIFO 调度程序。但是，为了允许多个并发用户，您可以控制每个应用程序将使用的最大资源数量。 默认情况下，它将获取集群中的所有核心，这仅在您一次只运行一个应用程序时才有意义。 您可以通过在 SparkConf 中设置 spark.cores.max 来限制核心数。 例如

val conf = new SparkConf()
  .setMaster(...)
  .setAppName(...)
  .set("spark.cores.max", "10")
val sc = new SparkContext(conf)

此外，您可以在集群 Master 进程上配置 spark.deploy.defaultCores，以将未设置 spark.cores.max 的应用程序的默认值更改为小于无限的值。 通过将以下内容添加到 conf/spark-env.sh 来实现

export SPARK_MASTER_OPTS="-Dspark.deploy.defaultCores=<value>"

这在共享集群上很有用，在这些集群上，用户可能没有单独配置最大核心数。

Executor 调度

分配给每个 Executor 的核心数量是可配置的。 当显式设置 spark.executor.cores 时，如果 Worker 具有足够的内核和内存，则可以在同一 Worker 上启动来自同一应用程序的多个 Executor。 否则，默认情况下，每个 Executor 都会获取 Worker 上所有可用的内核，在这种情况下，在单个调度迭代期间，每个应用程序只能在每个 Worker 上启动一个 Executor。

Stage 级别调度概述

Standalone 支持 Stage 级别的调度

• 当禁用动态分配时：它允许用户在 Stage 级别指定不同的任务资源需求，并将使用启动时请求的相同 Executor。
• 当启用动态分配时：目前，当 Master 为一个应用程序分配 Executor 时，它将根据多个 ResourceProfile 的 ResourceProfile ID 的顺序进行调度。具有较小 ID 的 ResourceProfile 将首先被调度。通常这无关紧要，因为 Spark 会在启动另一个 Stage 之前完成一个 Stage，唯一可能产生影响的情况是在作业服务器类型的场景中，因此需要记住这一点。 对于调度，我们将仅从内置的 Executor 资源中获取 Executor 内存和 Executor 内核，并从 ResourceProfile 中获取所有其他自定义资源，其他内置的 Executor 资源（例如 offHeap 和 memoryOverhead）将不起作用。 基础默认 Profile 将在您提交应用程序时根据 Spark 配置创建。 来自基础默认 Profile 的 Executor 内存和 Executor 内核可以传播到自定义 ResourceProfile，但所有其他自定义资源都不能传播。

注意事项

正如动态资源分配中所述，如果启用了动态分配但未明确指定每个 Executor 的内核数，则 Spark 可能会获取比预期更多的 Executor。 因此，建议在使用 Stage 级别调度时，为每个 ResourceProfile 显式设置 Executor 内核。

监控和日志记录

Spark 的独立模式提供了一个基于 Web 的用户界面来监控集群。Master 节点和每个 Worker 节点都有自己的 Web UI，显示集群和作业统计信息。默认情况下，您可以通过端口 8080 访问 Master 节点的 Web UI。该端口可以在配置文件中更改，也可以通过命令行选项更改。

此外，每个作业的详细日志输出也会写入每个 Worker 节点的工作目录（默认情况下是 SPARK_HOME/work）。对于每个作业，您将看到两个文件：stdout 和 stderr，其中包含其写入控制台的所有输出。

与 Hadoop 并行运行

您可以将 Spark 与现有的 Hadoop 集群一起运行，只需在相同的机器上将其作为单独的服务启动即可。要从 Spark 访问 Hadoop 数据，只需使用 hdfs:// URL（通常是 hdfs://<namenode>:9000/path，但您可以在 Hadoop Namenode 的 Web UI 上找到正确的 URL）。或者，您可以为 Spark 设置一个单独的集群，并仍然通过网络访问 HDFS；这比磁盘本地访问慢，但如果您仍然在同一个局域网中运行（例如，您在每个 Hadoop 机架上放置一些 Spark 机器），这可能不是问题。

为网络安全配置端口

一般来说，Spark 集群及其服务不会部署在公共互联网上。它们通常是私有服务，应仅限于需要访问这些服务的组织网络内访问。对 Spark 服务使用的主机和端口的访问应仅限于需要访问这些服务的源主机。

对于使用独立资源管理器的集群来说，这一点尤其重要，因为它们不像其他资源管理器那样支持细粒度的访问控制。

有关要配置的完整端口列表，请参阅安全页面。

高可用性

默认情况下，独立调度集群可以容忍 Worker 节点故障（因为 Spark 本身可以通过将工作移动到其他 Worker 节点来容忍工作丢失）。但是，调度器使用 Master 节点来做出调度决策，这（默认情况下）会创建一个单点故障：如果 Master 节点崩溃，则无法创建新的应用程序。为了规避这个问题，我们有两种高可用性方案，详见下文。

使用 ZooKeeper 的备用 Master

概述

利用 ZooKeeper 提供领导者选举和一些状态存储，您可以在集群中启动多个连接到同一 ZooKeeper 实例的 Master 节点。其中一个将被选为“领导者”，其余的将保持在备用模式。如果当前的领导者死亡，则将选举另一个 Master 节点，恢复旧 Master 节点的状态，然后恢复调度。整个恢复过程（从第一个领导者关闭时开始）应在 1 到 2 分钟之间。请注意，此延迟只会影响调度新应用程序 - 在 Master 节点故障转移期间已经运行的应用程序不受影响。

在此处了解有关 ZooKeeper 入门的更多信息 这里。

配置

为了启用此恢复模式，您可以通过配置 spark.deploy.recoveryMode 和相关的 spark.deploy.zookeeper.* 配置在 spark-env 中设置 SPARK_DAEMON_JAVA_OPTS。 有关这些配置的更多信息，请参阅 配置文档。

可能遇到的问题：如果您的集群中有多个 Master 节点，但未能正确配置 Master 节点以使用 ZooKeeper，则 Master 节点将无法相互发现，并认为它们都是领导者。这不会导致健康的集群状态（因为所有 Master 节点都会独立调度）。

详细信息

设置好 ZooKeeper 集群后，启用高可用性非常简单。只需使用相同的 ZooKeeper 配置（ZooKeeper URL 和目录）在不同的节点上启动多个 Master 节点进程。可以随时添加和删除 Master 节点。

为了调度新的应用程序或将 Worker 节点添加到集群，它们需要知道当前领导者的 IP 地址。这可以通过传入一个 Master 节点列表来实现，而您过去只传入一个 Master 节点。例如，您可以将您的 SparkContext 指向 spark://host1:port1,host2:port2。这将导致您的 SparkContext 尝试向两个 Master 节点注册 – 如果 host1 发生故障，此配置仍然正确，因为我们会找到新的领导者 host2。

“向 Master 节点注册”和正常操作之间存在一个重要的区别。启动时，应用程序或 Worker 节点需要能够找到并注册当前的领导 Master 节点。但是，一旦成功注册，它就“进入系统”（即存储在 ZooKeeper 中）。如果发生故障转移，新的领导者将联系所有先前注册的应用程序和 Worker 节点，通知他们领导地位的变化，因此他们甚至不需要知道新 Master 节点在启动时存在。

由于此属性，可以随时创建新的 Master 节点，您唯一需要担心的是新应用程序和 Worker 节点可以找到它以进行注册，以防它成为领导者。一旦注册，您就不用担心了。

使用本地文件系统的单节点恢复

概述

ZooKeeper 是生产级高可用性的最佳方式，但如果您只是想能够在 Master 节点关闭时重新启动它，FILESYSTEM 模式可以处理它。当应用程序和 Worker 节点注册时，它们会将足够的状态写入提供的目录，以便在 Master 节点进程重新启动时可以恢复它们。

配置

为了启用此恢复模式，您可以使用此配置在 spark-env 中设置 SPARK_DAEMON_JAVA_OPTS

详细信息

• 此解决方案可以与进程监视器/管理器（如 monit）结合使用，或者只是通过重新启动来启用手动恢复。
• 虽然文件系统恢复看起来比不进行任何恢复要好，但对于某些开发或实验目的来说，此模式可能不是最佳的。特别是，通过 stop-master.sh 杀死 Master 节点不会清除其恢复状态，因此每当您启动新的 Master 节点时，它都会进入恢复模式。如果它需要等待所有先前注册的 Worker 节点/客户端超时，这可能会使启动时间增加最多 1 分钟。
• 虽然未正式支持，但您可以将 NFS 目录挂载为恢复目录。如果原始 Master 节点完全死亡，您可以然后在不同的节点上启动一个 Master 节点，该节点将正确恢复所有先前注册的 Worker 节点/应用程序（相当于 ZooKeeper 恢复）。但是，未来的应用程序必须能够找到新的 Master 节点才能注册。