Apache Sparkのコンポーネントを整理する

# 背景・目的
あらためて、Apache Sparkのアーキテクチャやコンポーネントを整理します。

# 概要
## Components
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/c0e1b27f-6990-dfc8-24ea-46d5f441e568.png)
出典：[Apache Spark Cluster Mode Overview](https://spark.apache.org/docs/3.3.1/cluster-overview.html#components)

Sparkで利用される用語は下記の通りです。

| 用語 | 意味 |
|:-- | :--|
| Application | Spark上に構築されたユーザプログラム<BR>Driver ProcessとExecutor Processで構成される。 |
| Application jar |ユーザのSparkアプリケーションを含んだjarファイル |
| Driver program |main()関数を実行する。下記の役割を持つ。<BR>・ SparkContextを作成する<BR>・Sparkアプリケーションに関する情報管理<BR>・ユーザプログラムや入力の応答<BR>Executor全体のタスクの分析、分散、スケジューリングを行う。 |
| Cluster manager | リソースを取得するための外部プログラム（YARN、Mesos、k8s等）<BR>物理マシンを制御し、Sparkアプリケーションにリソースをアサインする。|
| Deploy mode| クラスターモードと、クライアントモードの2つがある。<BR>・クラスタモードの場合、Driverプログラムはクラスタ内で起動される。<BR>・クライアントモードの場合、Driverはクラスタ外から起動される。|
| Worker node | クラスタ内でアプリケーションを実行できる任意のノード|
| Executor |ワーカーノード上のアプリケーション用に起動されるプロセス。下記の役割がある。<BR>・Driverから割り当てたタスクを実行し、タスク全体のメモリまたはディスクストレージにデータを保持する。<BR>・Executorの計算の進捗をDriverにレポートする。 |
| Task |1つのエグゼキュータに送信される作業の単位 |
| Job |Sparkアクション (保存、収集など) に応答して生成される複数のタスクで構成される並列計算。|
| Stage |各ジョブは、互いに依存するステージと呼ばれる小さなタスクのセットに分割される。 (MapReduce の map ステージと reduce ステージに近いイメージ)。|

### Application、Job、Stage、Taskの関係性
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/36ad82f3-04fc-f214-3b33-11b505e611e4.png)

## 各Componentの解説
### Driver 
- Driverは、作成するプログラムのmain（）関数を実行するプロセス
- このプロセスが実行するユーザコードが、SparkContextの生成、RDDの生成を行い、変換やアクションを実行する
- ユーザプログラムのタスクへの変換を行う
    - Sparkドライバは、ユーザのプログラムを実行の単位（タスク）に変換する。
    - Sparkプログラムは、論理的なDAGを生成し、ドライバが実行されると、物理的な実行計画に変換する。
    - Sparkは、mapの変換を纏めてマージするためのパイプライン化のような最適化を行い、実行グラフを一連のステージ群に変換する。それぞれのステージには複数のタスクが含まれる。
    - タスクはSparkで最も小さな単位
- エグゼキュータ上のタスクのスケジューリングを行う
    - 物理的な実行計画が決まった後、Sparkドライバは、Executor群の個々のタスクのスケジュールを調整する。
    - Executor群は、起動時に自分自身をドライバに対して登録することで、ドライバは自分のアプリケーションのExecutorの状況を常に把握できる。
    - 各Executorは、タスクを実行し、RDDのデータを保持する事ができるプロセス。
    - Sparkドライバは、その時点でのExecutor群を見て、データの配置に基づいて各タスクを適切な場所にスケジューリングしようとする。
        - キャッシュされていた場合も、そのデータにアクセスする将来のタスクのスケジューリングに、この情報を活用する。
    - ドライバは、Webインタフェイスを通じて実行中のSparkアプリに関する情報を公開する

### Executor
- Sparkジョブの個々のタスクの実行を受け持つワーカープロセス
- ExecutorはSparkアプリケーションの起動時に1度起動されて、通常はそのアプリケーションが動作している間、動作し続ける。
- アプリケーションを構成するタスク群を実行し、結果をドライバに返す。
- ユーザプログラムによってキャッシュされるRDDをのためのインメモリストレージを各Executor内で動作するブロックマネージャーと呼ばれるサービスを通じて提供する。
    - RDDはExecutor内にキャッシュされるので、タスクはキャッシュされたデータと同居して動作することができる

### Cluster Manager
- ExecutorやDriverの起動をCluster Managerが担う。
- SparkのCluster Managerはプラガブルな構成となっており、YARNやMesosなどが利用できる。

#### Hadoop YARN
- 共有リソースプール上で様々なデータ処理フレームワークを動作させることができる。
- YARNで動作させる場合に、Sparkのアプリケーションは決められた数のExecutorを使う。各種パラメータは下記の通り

| 設定項目 | 説明 |
|:--|:--|
| --num-executors | Executorの数|
| --executor-memory | Executorが使用するメモリ |
| --executor-cors| YARNに要求するコア数|

### Spark Application
- Sparkアプリケーションの中核となるのは、Spark Driverとなる。
- Spark DriverではSparkSessionを作成する

### Spark Jobs
- DriverはSparkアプリケーションを1つ以上のSparkジョブに変換する。
- 次に各Spark ジョブをDAGに変換する。

### Spark Stages
- DAGノードの一部として、直列または並列に実行できる操作に基づいてステージが作成される。
- すべての操作が1つのStageになるわけではなく、複数のStageで分割できる。

### Spark Task
- 各ステージは、Sparkタスクで構成され、各Spark Executor間でフェデレーションされる。
- 各タスクは1つのコアにマップされる。
- 16コアのExecutorでは、16個以上のタスクを16個以上のパーティションで並列で処理する。

## 実行時の流れ
Driver プログラムのSparkContextが調整（コントロール）する。ざっくりと下記の流れになる。

1. SparkContextは、Cluster Managerにアクセスしてリソースを割り当てる。
1. 接続後、クラスタ内のノードでExecutorを取得する。
1. SparkContextから各Executorにアプリケーション（JarファイルやPythonファイル）を配布する。
1. SparkContextはタスクを各Executorに送信する。

## 留意事項
- Driverプログラムは、Executorからの通信をListenする必要があり、ワーカーノードからNWでアドレスを指定できる必要がある。
- Driverプログラムは、Workerノードを制御するため可能な限り近くで実行することが推奨される。

# 考察
基本的なところは整理できたので、今後はSparkの特徴である遅延評価実行等やパラメータについて整理しようとおもいます。

# 参考
https://spark.apache.org/docs/3.3.1/cluster-overview.html

https://spark.apache.org/docs/3.3.1/running-on-yarn.html#running-spark-on-yarn

https://www.databricks.com/jp/glossary/what-are-spark-applications