RDD の裏で DAG はどのように機能するのか? 質問する

私も、Spark が RDD から DAG を計算し、その後タスクを実行する方法について Web で調べていました。

大まかに言えば、RDD で何らかのアクションが呼び出されると、Spark は DAG を作成し、それを DAG スケジューラに送信します。

• DAG スケジューラは、演算子をタスクのステージに分割します。ステージは、入力データのパーティションに基づくタスクで構成されます。DAG スケジューラは、演算子をパイプライン化します。たとえば、多数のマップ演算子を 1 つのステージでスケジュールできます。DAG スケジューラの最終結果は、ステージのセットです。

• ステージはタスク スケジューラに渡されます。タスク スケジューラは、クラスター マネージャー (Spark Standalone/Yarn/Mesos) を介してタスクを起動します。タスク スケジューラは、ステージの依存関係を認識しません。

• ワーカーはスレーブ上でタスクを実行します。

Spark が DAG を構築する方法について見てみましょう。

高レベルでは、RDDに適用できる変換は2つあります。狭い変換と広い変換基本的に、ワイド変換はステージ境界をもたらします。

狭い変換- パーティション間でデータをシャッフルする必要はありません。たとえば、マップ、フィルターなど。

幅広い変革- たとえば、reduceByKey など、データをシャッフルする必要があります。

重大度レベルごとにログメッセージがいくつ表示されるかを数える例を見てみましょう。

以下は重大度レベルで始まるログファイルです。

INFO I'm Info message
WARN I'm a Warn message
INFO I'm another Info message

val input = sc.textFile("log.txt")
val splitedLines = input.map(line => line.split(" "))
                        .map(words => (words(0), 1))
                        .reduceByKey{(a,b) => a + b}

(2) ShuffledRDD[6] at reduceByKey at <console>:25 []
    +-(2) MapPartitionsRDD[5] at map at <console>:24 []
    |  MapPartitionsRDD[4] at map at <console>:23 []
    |  log.txt MapPartitionsRDD[1] at textFile at <console>:21 []
    |  log.txt HadoopRDD[0] at textFile at <console>:21 []