Spark スタンドアロンモード

• セキュリティ
• Spark スタンドアロンモードをクラスターにインストールする
• クラスターを手動で起動する
• クラスター起動スクリプト
• リソース割り当てと設定の概要
• アプリケーションをクラスターに接続する
• クライアントプロパティ
• Spark アプリケーションの起動
  • Spark プロトコル
  • REST API
• リソーススケジューリング
• Executor スケジューリング
• ステージレベルのスケジューリング概要
  • 注意点
• 監視とロギング
• Hadoop と並行して実行する
• ネットワークセキュリティのためのポート設定
• 高可用性
  • ZooKeeper によるスタンバイマスター
  • ファイルシステムによる単一ノードリカバリ

YARN クラスターマネージャー上で実行することに加えて、Spark はシンプルなスタンドアロンデプロイモードも提供します。スタンドアロンクラスターは、マスターとワーカーを手動で起動して手動で起動することも、提供されている 起動スクリプト を使用することもできます。テストのためにこれらのデーモンを 1 台のマシンで実行することも可能です。

セキュリティ

認証のようなセキュリティ機能はデフォルトでは有効になっていません。インターネットまたは信頼されていないネットワークに公開されているクラスターをデプロイする場合、不正なアプリケーションがクラスター上で実行されるのを防ぐために、クラスターへのアクセスを保護することが重要です。Spark を実行する前に、Spark セキュリティ およびこのドキュメントの特定のセキュリティセクションを参照してください。

Spark スタンドアロンモードをクラスターにインストールする

Spark スタンドアロンモードをインストールするには、コンパイル済みの Spark をクラスターの各ノードに配置するだけです。各リリースで事前ビルドされた Spark を入手することも、自分でビルド することもできます。

クラスターを手動で起動する

スタンドアロンマスターサーバーを起動するには、以下を実行します。

./sbin/start-master.sh

起動後、マスターは自身の spark://HOST:PORT URL を出力します。この URL を使用してワーカーを接続したり、SparkContext の「master」引数として渡したりできます。この URL は、マスターの Web UI (デフォルトでは https://:8080) でも確認できます。

同様に、1 つ以上のワーカーを起動し、以下を介してマスターに接続できます。

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

ワーカーを起動したら、マスターの Web UI (デフォルトでは https://:8080) を確認してください。新しいノードが、CPU 数とメモリ (OS 用に 1GB 残す) とともにリストされているはずです。

最後に、以下の設定オプションをマスターとワーカーに渡すことができます。

クラスター起動スクリプト

起動スクリプトを使用して Spark スタンドアロンクラスターを起動するには、Spark ディレクトリに conf/workers という名前のファイルを作成する必要があります。このファイルには、Spark ワーカーを起動する予定のすべてのマシンのホスト名を 1 行に 1 つずつ記述する必要があります。conf/workers が存在しない場合、起動スクリプトはデフォルトで単一マシン (localhost) を使用します。これはテストに便利です。注意、マスターマシンは各ワーカーマシンに SSH でアクセスします。デフォルトでは、SSH は並列で実行され、パスワードなし (秘密鍵を使用) のアクセス設定が必要です。パスワードなしの設定がない場合は、環境変数 SPARK_SSH_FOREGROUND を設定して、各ワーカーのパスワードをシリアルに入力できます。

このファイルをセットアップしたら、Hadoop のデプロイ スクリプトに基づいた以下のシェルスクリプトを使用して、クラスターを起動または停止できます。これらのスクリプトは SPARK_HOME/sbin にあります。

• sbin/start-master.sh - スクリプトが実行されたマシンでマスターインスタンスを起動します。
• sbin/start-workers.sh - conf/workers ファイルで指定された各マシンでワーカーインスタンスを起動します。
• sbin/start-worker.sh - スクリプトが実行されたマシンでワーカーインスタンスを起動します。
• sbin/start-connect-server.sh - スクリプトが実行されたマシンで Spark Connect サーバーを起動します。
• sbin/start-all.sh - 上記で説明したように、マスターと複数のワーカーを起動します。
• sbin/stop-master.sh - sbin/start-master.sh スクリプトで起動されたマスターを停止します。
• sbin/stop-worker.sh - スクリプトが実行されたマシンで、すべてのワーカーインスタンスを停止します。
• sbin/stop-workers.sh - conf/workers ファイルで指定されたマシンで、すべてのワーカーインスタンスを停止します。
• sbin/stop-connect-server.sh - スクリプトが実行されたマシンで、すべての Spark Connect サーバーインスタンスを停止します。
• sbin/stop-all.sh - 上記で説明したように、マスターとワーカーの両方を停止します。

これらのスクリプトは、ローカルマシンではなく、Spark マスターを実行したいマシンで実行する必要があることに注意してください。

conf/spark-env.sh で環境変数を設定することにより、クラスターをさらに設定することができます。このファイルは conf/spark-env.sh.template から開始して作成し、設定を有効にするために *すべてのワーカーマシンにコピー* してください。以下の設定が利用可能です。

注意: 起動スクリプトは現在 Windows をサポートしていません。Windows 上で Spark クラスターを実行するには、マスターとワーカーを手動で起動してください。

SPARK_MASTER_OPTS は以下のシステムプロパティをサポートしています。

SPARK_WORKER_OPTS は以下のシステムプロパティをサポートしています。

リソース割り当てと設定の概要

設定ページ のカスタムリソーススケジューリングと設定の概要セクションを必ずお読みください。このセクションでは、リソーススケジューリングの Spark スタンドアロン固有の側面のみについて説明します。

Spark スタンドアロンには 2 つの部分があります。1 つ目はワーカーのリソース設定、2 つ目は特定のアプリケーションのリソース割り当てです。

ユーザーは、割り当てるリソースセットを持つようにワーカーを設定する必要があります。spark.worker.resource.{resourceName}.amount を使用して、ワーカーが持つ各リソースの量を制御します。ユーザーは、ワーカーが割り当てられたリソースをどのように検出するかを指定するために、spark.worker.resourcesFile または spark.worker.resource.{resourceName}.discoveryScript のいずれかを指定する必要があります。どちらの方法がセットアップに適しているかについては、それぞれの上記の説明を参照してください。

2 番目の部分は、Spark スタンドアロンでアプリケーションを実行することです。標準の Spark リソース設定からの唯一の特別なケースは、ドライバをクライアントモードで実行する場合です。クライアントモードのドライバの場合、ユーザーは spark.driver.resourcesFile または spark.driver.resource.{resourceName}.discoveryScript を介して使用するリソースを指定できます。ドライバが他のドライバと同じホストで実行されている場合は、リソースファイルまたは検出スクリプトが同じノードで実行されている他のドライバと競合しないリソースのみを返すようにしてください。

注意、ワーカーはアプリケーションの各 Executor を割り当てられたリソースで起動するため、アプリケーションを送信するときにユーザーが検出スクリプトを指定する必要はありません。

アプリケーションをクラスターに接続する

Spark クラスターでアプリケーションを実行するには、マスターの spark://IP:PORT URL を SparkContext コンストラクタ に渡すだけです。

クラスターに対してインタラクティブな Spark シェルを実行するには、次のコマンドを実行します。

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

また、`--total-executor-cores <numCores>` オプションを渡して、spark-shell がクラスターで使用するコア数を制御することもできます。

クライアントプロパティ

Spark アプリケーションは、スタンドアロンモードに固有の以下の設定プロパティをサポートしています。

Spark アプリケーションの起動

Spark プロトコル

spark-submit スクリプト は、コンパイル済みの Spark アプリケーションをクラスターに送信する最も簡単な方法です。スタンドアロンクラスターの場合、Spark は現在 2 つのデプロイモードをサポートしています。`client` モードでは、ドライバはアプリケーションを送信するクライアントと同じプロセスで起動されます。しかし、`cluster` モードでは、ドライバはクラスター内のワーカープロセスのいずれかから起動され、クライアントプロセスはアプリケーションの完了を待たずに、アプリケーションを送信するという責任を果たしたらすぐに終了します。

アプリケーションが Spark submit 経由で起動された場合、アプリケーション JAR はすべてのワーカーノードに自動的に配布されます。アプリケーションが依存する追加の JAR については、カンマ区切り (例: `--jars jar1,jar2`) を使用して `--jars` フラグで指定する必要があります。アプリケーションの設定または実行環境を制御するには、Spark 設定 を参照してください。

さらに、スタンドアロンの `cluster` モードでは、アプリケーションがゼロ以外の終了コードで終了した場合に自動的に再起動できます。この機能を使用するには、アプリケーションを起動する際に `spark-submit` に `--supervise` フラグを渡すことができます。次に、繰り返し失敗するアプリケーションを終了したい場合は、以下を通じて行うことができます。

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

スタンドアロンマスターの Web UI (http://<master url>:8080) でドライバー ID を見つけることができます。

REST API

spark.master.rest.enabled が有効になっている場合、Spark マスターは http://[host:port]/[version]/submissions/[action] 経由で追加の REST API を提供します。ここで、host はマスターホスト、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" : "4.0.0",
  "submissionId" : "driver-20231124153531-0000",
  "success" : true
}

sparkProperties および environmentVariables の場合、ユーザーはサーバーサイド環境変数のプレースホルダーを次のように使用できます。

$ curl -XPOST http://IP:PORT/v1/submissions/create \
--header "Authorization: Bearer USER-PROVIDED-WEB-TOEN-SIGNED-BY-THE-SAME-SHARED-KEY"
...

sparkProperties および environmentVariables の場合、ユーザーはサーバーサイド環境変数のプレースホルダーを次のように使用できます。

...
  "sparkProperties": {
    "spark.hadoop.fs.s3a.endpoint": "{{AWS_ENDPOINT_URL}}",
    "spark.hadoop.fs.s3a.endpoint.region": "{{AWS_REGION}}"
  },
  "environmentVariables": {
    "AWS_CA_BUNDLE": "{{AWS_CA_BUNDLE}}"
  },
...

リソーススケジューリング

スタンドアロンクラスターモードは現在、アプリケーション間の単純な FIFO スケジューラのみをサポートしています。ただし、複数のユーザーが同時に実行できるように、各アプリケーションが使用する最大リソースを制御できます。デフォルトでは、クラスターの*すべて*のコアを取得しますが、これは一度に 1 つのアプリケーションのみを実行する場合にのみ意味があります。SparkConf の `spark.cores.max` を設定することで、コア数を制限できます。例:

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

さらに、クラスターマスタープロセスで `spark.deploy.defaultCores` を構成して、`spark.cores.max` を無限未満に設定していないアプリケーションのデフォルトを変更できます。これは、`conf/spark-env.sh` に以下を追加することで行います。

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

これは、ユーザーが個別に最大コア数を構成していない共有クラスターで役立ちます。

Executor スケジューリング

各 Executor に割り当てられるコア数は構成可能です。`spark.executor.cores` が明示的に設定されている場合、ワーカーに十分なコアとメモリがある場合、同じアプリケーションの複数の Executor が同じワーカーに起動される可能性があります。それ以外の場合、各 Executor はデフォルトでワーカーで利用可能なすべてのコアを取得するため、この場合、1 回のスケジューリングイテレーション中に各ワーカーに起動できるアプリケーションごとの Executor は 1 つだけになります。

ステージレベルのスケジューリング概要

ステージレベルのスケジューリングはスタンドアロンでサポートされています。

• 動的割り当てが無効な場合: ユーザーはステージレベルで異なるタスクリソース要件を指定でき、起動時に要求されたのと同じ Executor を使用します。
• 動的割り当てが有効な場合: 現在、マスターがアプリケーションの Executor を割り当てる場合、複数の ResourceProfile の ResourceProfile ID の順序に基づいてスケジュールされます。ID の小さい ResourceProfile が最初にスケジュールされます。通常、Spark は 1 つのステージを完了してから別のステージを開始するため、これは問題になりませんが、ジョブサーバータイプのシナリオでは影響を与える可能性があります。これは考慮すべき点です。スケジューリングでは、Executor メモリと Executor コアのみが組み込み Executor リソースから取得され、他のすべてのカスタムリソースは ResourceProfile から取得されます。Executor のオーバーヘッドなどの他の組み込み Executor リソースは影響しません。ベースのデフォルトプロファイルは、アプリケーションを送信するときに Spark 設定に基づいて作成されます。ベースのデフォルトプロファイルからの Executor メモリと Executor コアはカスタム ResourceProfile に伝播できますが、他のすべてのカスタムリソースは伝播できません。

注意点

動的リソース割り当てで述べたように、動的割り当てが有効な場合に各 Executor のコアが明示的に指定されていない場合、Spark は予想よりも多くの Executor を取得する可能性があります。そのため、ステージレベルのスケジューリングを使用する場合は、各リソースプロファイルに対して Executor コアを明示的に設定することをお勧めします。

監視とロギング

Spark のスタンドアロンモードは、クラスターを監視するための Web ベースのユーザーインターフェイスを提供します。マスターと各ワーカーは、クラスターとジョブの統計情報を表示する独自の Web UI を備えています。デフォルトでは、ポート 8080 でマスターの Web UI にアクセスできます。ポートは、設定ファイルで変更するか、コマンドラインオプションで変更できます。

さらに、各ジョブの詳細なログ出力は、各ワーカーノードの作業ディレクトリ (デフォルトでは SPARK_HOME/work) にも書き込まれます。ジョブごとに、stdout と stderr の 2 つのファイルがあり、コンソールに出力されたすべての出力が含まれます。

Hadoop と並行して実行する

既存の Hadoop クラスターと並行して Spark を実行するには、同じマシンで別個のサービスとして起動するだけです。Spark から Hadoop データにアクセスするには、hdfs:// URL (通常は hdfs://<namenode>:9000/path) を使用します。または、Spark 用に別のクラスターをセットアップし、ネットワーク経由で HDFS にアクセスすることもできます。これはディスクローカルアクセスよりも遅くなりますが、同じローカルエリアネットワーク内で実行している場合 (例: Hadoop がある各ラックに Spark マシンを数台配置する) は問題にならないかもしれません。

ネットワークセキュリティのためのポート設定

一般的に、Spark クラスターとそのサービスは公開インターネット上にはデプロイされません。それらは一般的にプライベートサービスであり、Spark をデプロイした組織のネットワーク内でのみアクセス可能であるべきです。Spark サービスで使用されるホストとポートへのアクセスは、サービスにアクセスする必要があるオリジンホストに限定されるべきです。

これは、スタンドアロンリソースマネージャーを使用するクラスターにとって特に重要です。なぜなら、他のリソースマネージャーが行うようなきめ細かなアクセス制御をサポートしていないためです。

設定すべきポートの完全なリストについては、セキュリティページ を参照してください。

高可用性

デフォルトでは、スタンドアロンスケジューリングクラスターはワーカー障害に対して耐性があります (Spark 自体が作業を他のワーカーに移動して失うことに耐性がある限り)。しかし、スケジューラはスケジューリングの決定を行うためにマスターを使用しており、これは (デフォルトで) 単一障害点を作成します。マスターがクラッシュした場合、新しいアプリケーションを作成できません。これを回避するために、2 つの高可用性スキームを用意しました。以下に詳述します。

ZooKeeper によるスタンバイマスター

概要

ZooKeeper を使用してリーダー選出と一部の状態保存を提供することで、同じ ZooKeeper インスタンスに接続された複数のマスターをクラスターに起動できます。1 つが「リーダー」として選出され、残りはスタンバイモードになります。現在のリーダーがダウンした場合、別のマスターが選出され、古いマスターの状態を回復してからスケジューリングを再開します。全体の回復プロセス (最初のリーダーがダウンしてから) は 1 分から 2 分で完了するはずです。この遅延は *新しい* アプリケーションのスケジューリングにのみ影響することに注意してください。マスターのフェイルオーバー中にすでに実行されていたアプリケーションは影響を受けません。

ZooKeeper の開始方法については、こちら で詳細を確認してください。

設定

このリカバリモードを有効にするには、`spark.deploy.recoveryMode` および関連する `spark.deploy.zookeeper.*` 設定を構成して、spark-env で `SPARK_DAEMON_JAVA_OPTS` を設定できます。

注意点: クラスターに複数のマスターがあるのに、マスターが ZooKeeper を使用するように正しく設定されていない場合、マスターはお互いを検出できず、すべてがリーダーであると認識します。これは健全なクラスター状態にはなりません (すべてのマスターが個別にスケジュールするため)。

詳細

ZooKeeper クラスターをセットアップしたら、高可用性の有効化は簡単です。同じ ZooKeeper 設定 (ZooKeeper URL とディレクトリ) を持つ異なるノードで複数のマスタープロセスを起動するだけです。マスターはいつでも追加および削除できます。

新しいアプリケーションをスケジュールしたり、クラスターにワーカーを追加したりするには、以前は単一のマスターを渡していた場所にマスターの IP アドレスのリストを渡す必要があります。たとえば、SparkContext を `spark://host1:port1,host2:port2` にポイントして起動するかもしれません。これにより、SparkContext は両方のマスターに登録しようとします。host1 がダウンしても、新しいリーダーである host2 を見つけることができるため、この構成は引き続き有効です。

「マスターへの登録」と通常の操作の間には重要な区別があります。起動時、アプリケーションまたはワーカーは現在のリーダーマスターを見つけて登録できる必要があります。しかし、一度正常に登録されると、それは「システム内」(つまり、ZooKeeper に保存されている) になります。フェイルオーバーが発生した場合、新しいリーダーは以前に登録されたすべてのアプリケーションとワーカーに連絡してリーダーシップの変更を通知するため、起動時に新しいマスターの存在を知る必要さえありません。

このプロパティにより、新しいマスターはいつでも作成できます。心配する必要があるのは、*新しい*アプリケーションとワーカーがリーダーになった場合に登録できることだけです。一度登録されると、すべて完了です。

ファイルシステムによる単一ノードリカバリ

概要

ZooKeeper は本番レベルの高可用性にとって最良の方法ですが、マスターがダウンした場合に再起動できるだけであれば、FILESYSTEM モードで対応できます。アプリケーションとワーカーが登録する際、提供されたディレクトリに十分な状態が書き込まれるため、マスタープロセスの再起動時に回復できます。

設定

このリカバリモードを有効にするには、spark-env で `SPARK_DAEMON_JAVA_OPTS` を次の設定で使用して設定できます。

詳細

• このソリューションは、monit のようなプロセスモニター/マネージャーと組み合わせて使用することも、手動リカバリを有効にするためにも使用できます。
• ファイルシステムリカバリは、まったくリカバリしないよりも明らかに優れていますが、このモードは特定の開発または実験目的には最適ではない可能性があります。特に、`stop-master.sh` でマスターを停止してもリカバリ状態はクリーンアップされないため、新しいマスターを起動するたびにリカバリモードに入ります。これは、以前に登録されたすべてのワーカー/クライアントのタイムアウトを待つ必要がある場合、起動時間が最大 1 分増加する可能性があります。
• 公式にはサポートされていませんが、NFS ディレクトリをリカバリディレクトリとしてマウントできます。元のマスターノードが完全にダウンした場合、別のノードでマスターを起動できます。これにより、以前に登録されたすべてのワーカー/アプリケーションが正しく回復されます (ZooKeeper リカバリと同等)。ただし、将来のアプリケーションは登録できるように、新しいマスターを見つける必要があります。