Sparkスタンドアローンモード

• セキュリティ
• Sparkスタンドアローンをクラスタにインストール
• クラスタを手動で開始
• クラスタ起動スクリプト
• リソース割り当てと設定の概要
• アプリケーションをクラスタに接続
• Client Properties
• Sparkアプリケーションの起動
• リソースのスケジューリング
• Executorのスケジューリング
• 監視および記録
• Hadoopと一緒に実行
• ネットワークセキュリティのためのポートの設定
• 高可用性
  • ZooKeeperを使ったスタンバイマスター
  • ローカルファイルシステムを使ったシングルモードリカバリ

MesosあるいはYARNクラスタマネージャー上での実行に加えて、Sparkは単純なスタンドアローン デプロイモードも提供します。マスターおよびワーカーを手動で開始するか、あるいは提供された起動スクリプトを使用するかのどちらかで、スタンドアローンクラスタを手動で起動することができます。テストのために一つのマシーン上でそれらのデーモンを起動することもできます。

セキュリティ

認証のようなセキュリティ機能はデフォルトでは有効にされていません。インターネットや信頼できないネットワークに開かれたクラスタを配備する場合、認証されていないアプリケーションをクラスタ上で実行することを防ぐためにクラスタへのアクセスを安全にすることが重要です。Sparkを実行する前にSpark セキュリティとこのドキュメントの特定のセキュリティの章を見てください。

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

Sparkスタンドアローンモードをインストールするには、クラスタ上の各ノード上にコンパイルされたバージョンのSparkを単に配置します。各リリースのSparkのpre-buildバージョンを取得するか、あるいは自身でビルドします。

クラスタを手動で開始

以下を実行することでスタンドアローンマスターサーバを起動することができます:

./sbin/start-master.sh

一度開始すると、マスターは自身でspark://HOST:PORT URLを出力するでしょう。ワーカーがそれに接続するために使うか、SparkContextの"master"引数として渡すことができます。マスターweb UI上でこのURLを見つけることもできます。デフォルトではhttp://localhost:8080 です。

同じように、1つ以上のワーカーを開始し、それらを以下のようにしてマスターに接続することができます:

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

一度ワーカーを開始して、マスターのweb UIを見てください(デフォルトでは、http://localhost:8080)。CPUの数とメモリと共に新しいノードがリストされているのを見るでしょう。(OSのために1ギガバイトが残されます)。

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

クラスタ起動スクリプト

起動スクリプトを使ってSparkスタンドアローンクラスタを起動するには、Sparkディレクトリに conf/workers と呼ばれるファイルを生成する必要があります。これにはSpark workerを開始しようとする全てのマシーンのホスト名が1行ごとに1つずつ含まれていなければなりません。conf/workersが無い場合は、起動スクリプトは初期値の1つのマシーン(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-all.sh - 上で説明されたマスターと多数のワーカーの両方を開始する。
• sbin/stop-master.sh - sbin/start-master.sh スクリプトで開始されたマスターを停止する。
• sbin/stop-worker.sh - スクリプトが実行されたマシーン上で全てのワーカーインスタンスを停止する。
• sbin/stop-worker.sh - conf/workers ファイルで指定されたマシーン上の全てのワーカーインスタンスを停止する。
• sbin/stop-all.sh - 上で説明されたマスターとワーカーの両方を停止する。

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

conf/spark-env.sh内で環境変数を設定することで、更にクラスタを任意に設定することができます。conf/spark-env.sh.templateからこのファイルを生成し、設定を有効にするためにそれを全てのワーカーマシーンにコピーします。以下の設定が利用可能です:

注意: 起動スクリプトは現在のところWindowsをサポートしません。SparkクラスタをWindows上で動作するには、手動でマスターおよびワーカーを開始してください。

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

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

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

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

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

ユーザは、一連のリソースを利用できるようにワーカーを構成して、それらを executor に割り当てることができるようにする必要があります。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クラスタ上でアプリケーションを実行するには、SparkContext コンストラクタにするように、単純にマスターのspark://IP:PORT URL を渡します。

クラスタに対して対話的なSparkシェルを実行するには以下のコマンドを実行します:

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

sparkシェルがクラスタ内で使用するコアの数を制御するために、オプション--total-executor-cores <numCores> を渡すこともできます。

Client Properties

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

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

spark-submit スクリプト はコンパイルされたSparkアプリケーションをクラスタにサブミットする最も率直な方法です。スタンドアローンクラスタに関しては、Sparkは現在のところ二つのデプロイモードをサポートします:クライアント モードでは、ドライバは同じプロセス内でクライアントがアプリケーションをサブミットしたかのように起動します。クラスタ モードでは、ドライバはクラスタ内のワーカープロセスの一つから起動されますが、クライアントプロセスはアプリケーションが終了するのを待たずにアプリケーションをサブミットする責任を果たすとすぐに終了します。

アプリケーションがSparkサブミットを使って起動されると、アプリケーションのjarは自動的に全てのワーカーノードに分配されます。アプリケーションが依存するどのような追加のjarについても、カンマを区切り文字として使って --jars フラグを使ってそれらを指定しなければなりません。アプリケーションの設定あるいは実行の環境を制御するには、Spark 設定を見てください。

更に、スタンドアローンクラスタモードは、アプリケーションが0ではない終了コードで終了した場合に、自動でのアプリケーションの再起動をサポートします。この機能を使うには、アプリケーションを起動する時にspark-submitに--supervise フラグを渡します。次に、繰り返し失敗するアプリケーションをkillしたい場合は、以下のようにして行うことができます:

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

http://<master url>:8080のスタンドアローンマスター web UIを使ってドライバIDを見つける事ができます。

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

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

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

spark.cores.maxをinfinite未満の何かに設定していないアプリケーションのためのデフォルトを変更するために、更にクラスタマスタープロセス上のspark.deploy.defaultCoresを設定することができます。以下をconf/spark-env.shに追加することでこれをしてください:

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

これはユーザが個々にコアの最大数を設定していないかもしれない共有クラスタで有用かも知れません。

Executorのスケジューリング

各executorへ割り当てられるコアの数は設定可能です。spark.executor.cores が明示的に設定された場合、もしワーカーが十分なコアとメモリを持つ場合は、同じアプリケーションからの複数のexecutorが同じワーカー上で起動されるかもしれません。そうでなければ、各executorはデフォルトでワーカー上で利用可能な全てのコアを横取りします。その場合1つのスケジュールの繰り返しの間にアプリケーション毎に1つのexecutorが各ワーカー乗で起動されるかも知れません。

監視および記録

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

更に、各ジョブの詳細なログ出力も各workerノードの作業デイlレクトリ(デフォルトではSPARK_HOME/work)に書き込まれます。各ジョブごとに2つのファイル、stdout と stderr、があるでしょう。全ての出力はコンソールに書き込まれます。

Hadoopと一緒に実行

同じマシーン上で別個のサービスとして起動するだけで、Sparkと一緒に既存のHaddopクラスタを実行することができます。SparkからHadoopデータにアクセスするためには、単に hdfs:// URL (一般的には hdfs://<namenode>:9000/path、正しいURLはHadoopネームノードのweb UIで見つけることができます)を使ってください。別のやり方として、Sparkのための個別のクラスタをセットアップし、ネットワーク越しにHDFSにまだアクセスできるようにすることができます; これはディスクのローカルアクセスより遅いですが、もし同じローカルエリアネットワーク内でまだ実行している場合は重要ではありません (例えば、Hadoopがある各ラック上に2、3のSparkマシーンを配置する)。

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

一般的に言って、Spark クラスタとそのサービスは公的なインターネットに配備されません。それらは一般的にプライベートなサービスで、Sparkを配備する組織のネットワーク内でのみアクセス可能な筈です。Sparkサービスによって使われるホストとポートへのアクセスはサービスへアクセスする必要があるオリジンのホストに制限されるべきです。

スタンドアローン リソースマネージャーを使うクラスタは他のリソースマネージャーがする方法でfine-grainedアクセス制御をサポートしないため、これは特にそれにとって重要です。

設定するポートの完全なリストは、 セキュリティのページを見てください。

高可用性

デフォルトでは、スタンドアローンスケジューリングクラスタはワーカーの障害に対して回復力に富んでいます(Spark自体の範囲においてワークの喪失に対して他のワーカーにそれを移動することで回復力に富んでいます)。しかし、スケジューラーはスケジュールの決定をするためにマスターを使用します。これにより(デフォルト)単一障害点ができます: もしマスターがクラッシュすると新しいアプリケーションを生成することができません。これを回避するために、2つの高可用性の仕組みがあります。以下に詳細を示します。

ZooKeeperを使ったスタンバイマスター

概要

リーダーの選出と幾つかの状態のストレージを提供するためにZooKeeperを使用すると、同じZooKeeperインスタンスに接続しているクラスタ内に複数のマスターを起動することができます。一つが"リーダー"として選出され、他はスタンドバイモードのままでいるでしょう。現在のリーダーが死亡すると、他のマスターが選出され、古いマスターの状態を回復してスケジューリングを再開します。回復の全プロセス(最初のリーダーがダウンしてから)には、1,2分かかるでしょう。この遅延は新しい アプリケーションのスケジューリングにのみ影響することに注意してください - マスター障害の間に既に実行中のアプリケーションは影響を受けません。

ZooKeeperを使った開始についてもっと知るにはここを見てください。

設定

この回復モードを有効にするために、spark.deploy.recoveryMode と関係する spark.deploy.zookeeper.* 設定を設定することで、SPARK_DAEMON_JAVA_OPTS を設定することができます。これらの設定についての詳しい情報は設定ドキュメントを参照してください。

Possible gotcha: クラスタ内に複数のマスターがあるが、ZooKeeperを使うように正しくマスターを設定できていない場合、マスターはお互いを見つける事ができずにそれぞれが全てリーダーだと考えるかもしれません。これは結果として(全てのマスターが独立してスケジュールすることになるため)healthyクラスタ状態ではなくなるでしょう

詳細

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

新しいアプリケーションのスケジュールあるいはクラスタにワーカーを追加するためには、それらがクラスタのリーダーのIPアドレスを知る必要があります。一つの時に渡していたマスターのリストに単純に渡すことで行うことができます。例えば、spark://host1:port1,host2:port2を指しているSparkContextを開始しても構いません。これはSparkContextが両方のマスターを登録するようにさせるでしょう - もし host1がダウンすると、新しいリーダーhost2を見つけるのでこの設定はまだ正しいでしょう。

"マスターの登録"と通常の操作の間で重要な違いがあります。開始時にはアプリケーションあるいはワーカーは現在のリーダーマスターを見つけて登録できる必要があります。とはいえ、一旦登録に成功すると、それは"システム内"になります (つまり、ZooKeeperに格納されます)。障害が発生すると、新しいリーダーはリーダーシップの変更を全ての以前に登録されたアプリケーションとワーカーに知らせるために接続します。そのためそれらは起動時に新しいマスターの存在を知る必要さえありません。

この特性のため、新しいマスターはいつでも生成することができ、それがリーダーになった場合に新しいアプリケーションとワーカーがそれを登録するために見つけることができるかを心配するだけで良いです。一旦登録されると、ZooKeeperが面倒を見ます。

ローカルファイルシステムを使ったシングルモードリカバリ

概要

ZooKeeperはプロダクションレベルの高可用性を追求するための最も良い方法ですが、もしマスターがダウンしたら単に再起動できるようにしたいと思う場合は、FILESYSTEMモードがそれを引き受けます。アプリケーションおよびワーカーを登録する時に、それらはマスタープロセスが再起動する場合に回復できるように提供されたディレクトリに十分な状態を書き込みます。

設定

この回復モードを有効にするために、以下の設定を使ってspark-envの中でSPARK_DAEMON_JAVA_OPTSを設定することができます。

詳細

• この解決方法はmonitのような監視/管理プロセスと提携して使われるか、単に再起動することで手動のリカバリをするために使われます。
• ファイルシステムの回復は何も回復しないよりは素直に良いように思えますが、このモードでは特定の開発あるいは実験的な目的のためには最適状態には及ばないかも知れません。特に、stop-master.shを使ってマスターをkillすることはリカバリー状態を掃除しないため、新しいマスターを開始するときはいつでもリカバリーモードに入るでしょう。これは全ての以前の登録済みのワーカー/クライアントがタイムアウトするのを待つ必要がある場合には1分間開始の時間を増加させるでしょう。
• 公式にはサポートされていませんが、リカバリーディレクトリとしてNFSディレクトリをマウントすることができます。元のマスターノードが完全にダウンした場合、異なるノードでマスターを開始することができます。全ての以前の登録済みのワーカー/アプリケーションが正しくリカバーされるでしょう(ZooKeeperのリカバリーに相当します)。しかし、将来のアプリケーションは登録するために新しいマスターを見つけることができる必要があるでしょう。