一、概要 この記事では、wordCountを例にSparkがジョブを作成し実行するプロセスについて詳しく説明します。特に、プロセスとスレッドの作成に焦点を当てます。

二、実験環境の設定 以下の条件を満たしていることを確認してください。

1. Sparkバイナリ0.9.1をダウンロード
2. Scalaをインストール
3. sbtをインストール
4. Javaをインストール

三、spark-shellの起動 3.1 ローカルモードでの実行 ローカルモードでの実行は非常に簡単です。以下のコマンドを実行します（現在のディレクトリが$SPARK_HOMEであると仮定）。

MASTER=local bin/spark-shell

"MASTER=local"は単一マシンモードで実行することを示しています。

3.2 ローカルクラスターモードでの実行 ローカルクラスターモードは、単一マシン上でスタンドアロンクラスターを模倣する擬似クラスターモードです。以下の順序で起動します。

1. マスターの起動
2. ワーカーの起動
3. spark-shellの起動

3.3 マスターの起動

$SPARK_HOME/sbin/start-master.sh

ログはデフォルトで$SPARK_HOME/logsディレクトリに保存されます。 マスターは主にorg.apache.spark.deploy.master.Masterクラスを実行し、8080ポートでリッスンします。

3.4 設定の変更

1. $SPARK_HOME/confディレクトリに移動
2. spark-env.sh.templateをspark-env.shにリネーム
3. spark-env.shを編集し、以下の内容を追加

export SPARK_MASTER_IP=localhost
export SPARK_LOCAL_IP=localhost

3.5 ワーカーの実行

bin/spark-class org.apache.spark.deploy.worker.Worker spark://localhost:7077 -i 127.0.0.1 -c 1 -m 512M

ワーカーが起動し、マスターに接続します。マスターのWeb UI（http://localhost:8080）を開くと、接続されたワーカーを見ることができます。

3.6 spark-shellの起動

MASTER=spark://localhost:7077 bin/spark-shell

すべてが順調であれば、以下のメッセージが表示されます。

Created spark context..
Spark context available as sc.

ブラウザでlocalhost:4040を開くと、以下の情報を確認できます。

• ステージ
• ストレージ
• 環境
• エグゼキューター

四、wordcountの実行 環境設定が完了したら、最もシンプルな例としてwordcountを実行してみましょう。spark-shellで以下のコードを入力します。

scala> sc.textFile("README.md").filter(_.contains("Spark")).count

上記のコードは、README.mdファイルの中で"Spark"という文字列を含む行数をカウントします。

五、ジョブの生成と実行 ジョブ生成の基本的な流れは以下の通りです。

1. アプリケーションがSparkContextのインスタンスを作成します（例：sc）。
2. SparkContextを使ってRDDを生成します。
3. 一連のtransformation操作により、元のRDDが別のタイプのRDDに変換されます。
4. actionが変換後のRDDに適用されると、SparkContextのrunJobメソッドが呼び出されます。
5. **sc.runJobの呼び出し**は、その後の一連の反応の始まりであり、重要な変化がここで起こります。

呼び出しパスは以下のようになります。

1. sc.runJob -> dagScheduler.runJob -> submitJob
2. DAGScheduler::submitJobはJobSubmittedイベントを内部クラス**eventProcessActor**に送信します。
3. eventProcessActorはJobSubmittedイベントを受け取り、**processEvent**処理関数を呼び出します。
4. jobからstageへの変換を行い、finalStageを生成し実行します。鍵となるのは**submitStage**の呼び出しです。
5. submitStageでは、stage間の依存関係を計算します。依存関係には**幅依存**と**狭依存**があります。
6. 計算中に現在のstageが依存関係を持たないか、または全ての依存関係が準備できている場合、taskを提出します。
7. taskの提出は**submitMissingTasks**関数を呼び出すことで行われます。
8. taskがどのワーカー上で実行されるかはTaskSchedulerによって管理されます。つまり、submitMissingTasksはTaskScheduler::submitTasksを呼び出します。
9. TaskSchedulerImplは、Sparkの現在の実行モードに基づいて適切なbackendを作成します。単一マシンで実行されている場合はLocalBackendを作成します。
10. LocalBackendはTaskSchedulerImplから受け取った**ReceiveOffers**イベントを処理します。
11. receiveOffers -> executor.launchTask -> TaskRunner.run

executor.launchTaskのコードスニペット:

def launchTask(context: ExecutorBackend, taskId: Long, serializedTask: ByteBuffer) {
  val tr = new TaskRunner(context, taskId, serializedTask)
  runningTasks.put(taskId, tr)
  threadPool.execute(tr)
}

以上の説明から、最終的な論理処理はTaskRunnerというexecutor内で行われることがわかります。計算結果はMapStatusに包装され、一連の内部メッセージを通じてDAGSchedulerにフィードバックされます。