Sparkのアーキテクチャと処理の流れ

# はじめに
* Sparkの用語や処理の流れのメモです。

# Sparkとは
* `Apache Spark™ is a unified analytics engine for large-scale data processing.`
* 大規模データに対して高速に分散処理を行うオープンソースのフレームワーク。大量のメモリを活用して高速化を実現するのが特徴。

# Sparkの仕組み
## アーキテクチャ

![スクリーンショット 2021-08-18 14.38.33.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/92368/73e56718-0735-18b6-e1ff-4340233b00ff.png)


* Clientモードの概要図は、[Spark Architecture](https://www.learntospark.com/2020/02/spark-architecture.html)が分かりやすい

## 処理の流れ
### ①ClientがResource Managerに、DriverとExecutorの割り当てを依頼する
* Executorの数、要求するリソース量（コア数/メモリ量）などを指示する
* Driverの位置がURLで返される
* Clientモードの場合は、DriverプログラムがCluster Managerに対してResource Negotiationをかける

| 用語        | 意味                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              | 
| ----------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | 
| Application | ・Driverプログラム + Executor から構成される<br>・ApplicationごとにExecutorが起動され、共有されることはない                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                       | 
| Driver      | ・SparkContextオブジェクトを生成するコンポーネント（DataFrameやDataset APIの場合はSparkSession）<br>　・-> SparkContextがSparkアプリケーション全体を通して状態を管理して、様々なAPIをユーザーに提供する<br>・RDDの生成や変換の処理を記述したもの<br>・-> ユーザプログラムをTaskに変換し、Executorに実行を命令する + Actionに伴いExecutorからデータを取得する<br>・Executorは起動時に自分自身をDriverに対して登録するため、Driverは自分のApplicationのExecutorの様子を常に把握できる<br>・インタラクティブシェルの場合は起動時にClient側で生成されているので、明示的に用意しなくて良い<br>　・インタラクティブシェルであっても、クラスタ上で分散処理ができる<br>　・spark-shellとspark-submit<br>・JVM上で動作する | 
| Executor    | ・Driverから命令されたTaskを実行する<br>・Worker Nodeごとに1プロセス起動し、その中で複数のスレッドを起動する<br>・RDDをキャッシュする<br>・JVM上で動作する                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        | 

| 用語          | 意味                                                                                                                                                                                                     | 
| ------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | 
| Clientモード  | ・Client側でDriverプログラムが実行される                                                                                                                                                                 | 
| Clusterモード | ・Application Master上でDriverプログラムが実行される<br>（ApplicationがsubmitされてSparkContextが生成されると、SparkContextはCluster Managerに処理を実行するためのコンピュートリソースをリクエストする） | 

| 用語                           | 意味                                                                                             | 
| ------------------------------ | ------------------------------------------------------------------------------------------------ | 
| Resource Manager, Node Manager | ・Standalone, Yarn, Apache Mesos, Kubernetesから選択できる<br>・Standalone: Hadoopなしで動作する | 

### ②Worker Nodeにて、Driverを起動する

### ③Driverが、ユーザが記述したRDD変換手順からLineageを作成する
* ユーザプログラムがTransormationを組み合わせてRDD参照グラフを作成（③）
* -> ユーザプログラムがActionを実行
* -> DriverがRDD参照グラフからStageのDagを作成（④）
* -> Executorに各Stageの処理を命令する（⑦）
* -> Actionに伴いExecutorからデータを取得する（⑨）

### ④Lineageを元に、JobをTaskに分割し、個々のTaskをSckedulerに割り当てる

| 用語                    | 意味                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  | 
| ----------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | 
| RDD, Dataframe, Dataset | ・処理対象データセット。データコレクションの抽象表現。<br>・パーティションという単位に分割されてノードごとに分散配置される<br>・イミュータブル。再利用前提なので。処理の中で新しいRDDを作っていく。<br>　・キャッシュや複数ノードへのレプリケーションのため<br>　・イミュータブルでないと値が変更されるたびにキャッシュを複数ノード間で同期する必要があり、複雑な処理が必要になる<br>・入力データを読み込みRDD化して、RDDに対して次々に変換を行ったあと、最終的に結果を出力する<br>・インターフェイスは2つ<br>　・Transormation<br>　　・既存のRDDから、変換処理などを通じて新しいRDDを作り出す<br>　・Action<br>　　・Driverにデータを戻したり、外部にデータを書き込んだり<br>　　・Actionのタイミング（=実際にデータが必要になるタイミング）で遅延実行される<br>・PythonはDataframe | 
| Lineage                 | ・RDD参照グラフ<br>・LineageからDAGが生成される<br>・DAGをさかのぼることで同一のRDDを作り直すことができる<br>・Sparkは、データセットとしてのRDDとプログラミングモデルとしてのDAGの2つがコアアイデア+オンメモリでの分散処理に特化していること                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          | 
| Scheduler               | ・Dag SchedulerとTask Scheduler<br>　・Dag SchedulerがLineageをStageに分割する<br>　・-> パーティションごとにExecutor1つがまとめて計算できる変換の範囲を決めるため<br>・DAG: RDDを変換していく過程。論理実行プラン。<br>・JobをTaskに分割したり、TaskをExecutorに割り当てる                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                           | 

| 用語        | 意味                                                                                                                          | 
| ----------- | ----------------------------------------------------------------------------------------------------------------------------- | 
| Application | ・submitする単位                                                                                                              | 
| Job         | ・RDDの変換を繰り返して目的の結果を得る一連の処理<br>・独立した実行アクション<br>・Actionを実行するとJobが作成される          | 
| Stage       | ・JobはShuffleが必要となるタイミングで複数のStageを分ける<br>　・<- 別パーティションが必要となるタイミングでShuffleが発生する | 
| Task        | ・最小実行単位                                                                                                                | 

### ⑤Executorを起動するWorker Nodeを割り当てる

### ⑥Jobに割り当てられたExecutorをDriverに通知する

### ⑦Driverが、ExecutorにTaskを割り当てる
* アプリケーションコードをExecutorに送ってTaskを実行する

### ⑧シャッフルがある場合、Executor間で多対多通信をする

### ⑨Executorが、DriverにTaskの実行結果を通知する

# 参考
* [Apache Spark](https://spark.apache.org)
* [Spark Architecture](https://www.learntospark.com/2020/02/spark-architecture.html)
* [Apache Sparkについて](https://www.slideshare.net/BrainPad/spark-basic)
* [Apache Sparkのご紹介 ](https://www.slideshare.net/hadoopxnttdata/apache-spark)
* [Apache Sparkの概要](https://qiita.com/whata/items/8915182cbd3759eebe6d)
* [Apache Spark で分散処理入門](https://qiita.com/Hiroki11x/items/4f5129094da4c91955bc)
* [詳解 Apache Spark](https://www.amazon.co.jp/%E8%A9%B3%E8%A7%A3-Apache-Spark-%E4%B8%8B%E7%94%B0-%E5%80%AB%E5%A4%A7/dp/4774181242)