SparkとYARNについて書きます。テーマ的にインフラストラクチャについての話が多くなると思います。

SparkとHadoopの関係性

SparkはHadoopクラスタへの依存はしていない。(ただし、ややこしいのだがHDFSやYARNのクライアントライブラリへの依存はある)なのでHadoopなしでも動かすことができる。しかしそれでもHadoopと一緒に動作させることが多いのは以下の理由による。

クラスタマネージャとしてのYARN

Sparkはアプリケーション（厳密にはSparkアプリケーション）ごとに下記のようなクラスタが構築される。Driver Programと呼ばれる、SparkContextオブジェクトを持ち、アプリケーションコードの主要部分を実行するアプリケーションのマスタコンポーネントと、RDDに対するオペレーションを実行するExecutor群。そして、Driver Programからの要求に応じてExecutorをアロケーションするCluster Manager。

現在のSparkでは、Cluster ManagerとしてStandalone Cluster Manager(Spark自体が用意する仕組み）、MesosそしてYARNを利用することができる。YARNを利用する場合には以下のような構成になる。（Taskの表記がここではタスク、とカタカナになっているのはご愛嬌）

./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master yarn-client \
  --num-executoers 2 \
  /path/to/examples.jar

例えば上記のような形でアプリケーションを呼び出した場合、以下のような流れを辿ってアプリケーション用のSparkクラスタが構築される。(アプリケーションごとにこのクラスタが構築されるので、「クラスタ」という言葉は若干違和感がある）

クライアントマシン上でDriver Programが起動
Driver ProgramがYARNのResourceManager(以下RM)に対してYARNアプリケーションとしてSparkのjarをサブミット
起動してきたアプリケーションマスタがExecutorとして2つのタスクコンテナを要求
結果として上記のような図の構成になる。

なおmasterにyarn-clientを指定した場合、Driver Programはアプリケーション呼び出しを行ったクライアントマシン上で動作する。yarn-clusterを指定した場合はDriver Program自体もYARNのタスクコンテナ上で動作する。

というのがYARNとSparkのざっくりした関係性だ。

データソースとしてのHDFS

SparkはさまざまなストレージシステムをRDDのデータソースとして取り扱うことができる。HDFSもそのひとつであり、以下の様に簡単にHDFS上からRDDを生成することができる。

val rdd = sc.textFile("hdfs://path/to/your/data")

このtextFile()というメソッドはRDDを返すメソッドであり、呼びだされた時点ではHDFSへのReadは発生しない。RDDに対するアクションが呼び出されて各Executor上での処理が行われる際、最初のステージで各タスクからHDFSへのReadが発生することになる。

SparkはYARNで動かすべきか

HDFS上のデータに対する処理を行うような場合はYARNで動かすことのメリットが大きいと感じる。メリットが大きい、というか、HDFSがある場所にはだいたいYARNも動いていることが多いと思うので、既存の資産を活かしやすいという感じだろうか。また、下記のようにYARNがまさにリソースマネージャの役割を果たしてくれるので、アドホックなデータアプリケーションプラットフォーム的にマルチテナント化がしやすい。更にMapReduceやHiveなどの他のHadoopアプリケーションと共存させやすい、というメリットが見込める。

一方、特定の目的のためにサービスとしてSpark Streamingのアプリケーションを動作させるような場合など、マルチテナントでアプリケーションをアドホックに動かすことがないようなケースはYARNでなくてもよいのではないだろうか。

以下、YARNがSparkにもたらすメリットを紹介する。

Dynamic Resource Allocation

Spark1.2以降では、アプリケーションのリソースを動的に増減させることができるようになっており、具体的にはExecutor単位でのリソースの増減が行なわれる。現在のところYARNをクラスタマネージャとして利用したときのみにこの機能が利用可能。

まずspark.dynamicAllocationをtrueに設定する。更にExternal Shuffle Serviceと呼ばれるサービスを利用する必要がある。これはshuffleファイルをローカルストレージではなく外部に保存し、既存のExecutorを安全に削除するためのサービスだ。　spark.shuffle.service.enabledをtrueにすることによって、アプリケーションにこのサービスの利用を設定します。サービス自体はorg.apache.spark.yarn.network.YarnShuffleServiceというクラスに実装されており、各ワーカーノードのNodeManager内で動作する。起動するためにはspark-<version>-yarn-shuffle.jarをワーカーノードに配布し、さらにyarn-site.xml内のyarn.nodemanager.aux-services配下にyarn.nodemanager.axu-services.spark_shuffleというプロパティを作り、ここに先ほどのクラス名を指定する。その後NodeManagerを再起動してサービスを有効化することで利用可能になる。

この状態でアプリケーション起動後、spark.dynamicAllocation.schedulerBAcklogTimeoutの時間が過ぎるとspark.dynamicAllocationSchedulerBacklogTimeoutのインターバルで新しいExecutorの要求を開始される。初回は1つのExecutorを追加し、その後2、4、8とExponentialに追加数を増やしていく。そしてspark.dynamicAllocation.executorIdleTimeoutで指定された秒数のアイドル時間が続くとそのExecutorは破棄される。

YARNで強化するSparkのセキュリティ対策

Spark自体が提供するセキュリティ対策のほかに、YARNのCapacity Schedulerを使うことによって、RMに対してアプリケーション登録時に認証をかけることができるようになる。ちょっと時間切れになってきたので、詳細はまたあとで追記。

Spark自体のセキュリティについてはこちらも参照するとよさそう。
Secure Spark

まとめ

ということでSparkとYARN、というかHadoopの関係性や、それらがもたらしてくれるメリットの紹介でした！

Disclaimer

本記事はわたしの雇用者の意見を代表するものではありません。