![5Copyright ? 2013 NTT DATA Corporation
join
map
もう少し複雑な処理のイメージ
HDFS上のファイルやリスト
などのデータ構造など、
様々なデータソースから
RDDを生成できる
ListなどのScalaコ
レクションオブジェ
クト
HDFS
HDFSなど様々
なデータソース
からRDDを生
成可能
?
?
? Actionが適用された時点
でジョブが生成され、実際
の計算が始まる。
RDDはイミュータブルなの
で、このように遅延計算し
ても、逐次計算した場合
と同じ結果が得られる
filter
filter
distinct
textFile
parallelize
filter map
原則、関数評価の直
後に計算しない。生
成された子RDDには
親RDDと後に計算す
る関数を記憶してお
く。
このため計算コスト
やオブジェクトの生成
コストが削減される
map
filterdistinct
join
reduceByKey
計算結果
join
distinct
[関数]
関数評価後のRDDRDD
RDDの変換の方向
変換関数
Action
RDDはイミュータブル
なので、変換のもと
になったデータを変
更せず、新規にデー
タを生成する
最終的な計算結果](https://image.slidesharecdn.com/apache-spark-nttdata-2-140529210313-phpapp02/85/Apache-Spark-5-320.jpg)
More Related Content
What's hot (20)
Similar to Apache Sparkのご紹介 （後半：技術トピック） (20)
![[db tech showcase Tokyo 2014] D33: Prestoで実現するインタラクティブクエリ by トレジャーデータ株式会社 斉藤太郎](https://cdn.slidesharecdn.com/ss_thumbnails/dbtstokyo2014d33presto-141120012543-conversion-gate01-thumbnail.jpg?width=560&fit=bounds)
More from NTT DATA OSS Professional Services (20)
Apache Sparkのご紹介 （後半：技術トピック）
- 1. Apache Sparkのご紹介 ～技術トピック 2014年5月29日 第16回 Hadoopソースコードリーディング 発表資料 NTTデータ 基盤システム事業本部 OSSプロフェッショナルサービス 猿田 浩輔
- 2. 2Copyright ? 2013 NTT DATA Corporation ? 猿田 浩輔 ? Hadoopを中心とするOSSの技術検証や導入支援、テクニカ ルサポートを行う ? 最近はSparkに興味を持つ ? Hadoop徹底入門 / HADOOP HACKSを世に出す悪事に加 担 自己紹介
- 3. 3Copyright ? 2013 NTT DATA Corporation ? Sparkでの典型的な処理のイメージ ? Sparkが扱う基本的な抽象データセット「RDD」の概要 ? クラスタ上での処理や、ノード間の通信イメージ ? RDDの変換チェインがクラスタ上で実行可能な処理に分割されるまで の流れ ここからのトピック
- 4. 4Copyright ? 2013 NTT DATA Corporation ? SparkではRDDと呼ばれる抽象データセットの変換を繰り返して目的の結果を得る ? この一連の処理を、「ジョブ」と呼ぶ ? Spark版ワードカウントの例 典型的な処理イメージ // HDFS上のファイルからRDDを生成 1: val hdfsRdd = spark.textFile("hdfs://...") // 単語で構成されるシーケンスを作る 2: val splitRdd = hdfsRdd.flatMap(line => line.split(" ")) // 各単語をキーとし、値を1とするキーバリューペアを作る 3: val mappedRdd = mappedRdd.map(word => (word, 1)) // 単語をキーとし、同じキーを持つペアの値(1)を足しこむ 4: val reducedRdd = mappedRdd.reduceByKey(_ + _) // 計算結果をHDFS上のファイルに書き出す 5: counts.saveAsTextFile("hdfs://...")
- 5. 5Copyright ? 2013 NTT DATA Corporation join map もう少し複雑な処理のイメージ HDFS上のファイルやリスト などのデータ構造など、 様々なデータソースから RDDを生成できる ListなどのScalaコ レクションオブジェ クト HDFS HDFSなど様々 なデータソース からRDDを生 成可能 ? ? ? Actionが適用された時点 でジョブが生成され、実際 の計算が始まる。 RDDはイミュータブルなの で、このように遅延計算し ても、逐次計算した場合 と同じ結果が得られる filter filter distinct textFile parallelize filter map 原則、関数評価の直 後に計算しない。生 成された子RDDには 親RDDと後に計算す る関数を記憶してお く。 このため計算コスト やオブジェクトの生成 コストが削減される map filterdistinct join reduceByKey 計算結果 join distinct [関数] 関数評価後のRDDRDD RDDの変換の方向 変換関数 Action RDDはイミュータブル なので、変換のもと になったデータを変 更せず、新規にデー タを生成する 最終的な計算結果
- 6. 6Copyright ? 2013 NTT DATA Corporation Sparkのデータ処理をざっくり言うと ? RDD（Resilient Distributed Datasets）という 抽象データセットを繰り返し変換し、結果を得る ? ユーザが定義したRDDの系譜（Lineage）に従って 処理を実施する それで、RDDって？ RDDのつながりのグラフ構造
- 7. 7Copyright ? 2013NTT DATA Corporation RDDとは? ? コレクションのようなデータ構造であり、内部の要素をイテレートできる ? 更に、Sparkの基本的なコンセプトを実現するために、次のような特徴を備える ? パーティションに分割され、サーバ上で分散配置される ? インメモリで保持される(なるべくディスクに着地させない) ? 遅延計算される(計算を行うタイミングをスケジューリングされたあと、実際に計算される) ? イミュータブル(フォールトトレラント性を担保するための特徴) RDD 要素（レコード） 要素（レコード） パーティション 要素（レコード） 要素（レコード） パーティション RDDの基本構造 サーバ サーバ パーティションの単位で 複数のサーバに分散配置される。 パーティションの数、分割の方法はRDDの種類や 変換の種類によって決まる。 実際の計算は要素ごとに 実施される。
- 8. 8Copyright ? 2013 NTT DATA Corporation 遅延計算 val sc = new SparkContext(...) val file = sc.textFile("hdfs://...") val errors = file.filter(line => line.contains("ERROR")) // Count all the errors errors.count() この時点では計算されない ここで計算がスケジューリングされる。 実際に計算が起こるタイミングはスケジューリングによる。 サンプルコード 計算が スケジューリ ングされない 計算 がスケジュー リングされる
- 9. 9Copyright ? 2013 NTT DATA Corporation フォールトトレラント性を実現するための特性 ? Sparkでは、大量のデータをオンメモリで分散処理するので、各サーバに分散配置され たデータ(パーティション)が処理中に欠損した場合に備えた仕組みが必要 ? 欠損に備えつつ、ネットワーク転送を出来る限り避けたい ? 分散システムでは扱うデータをコピー（レプリカ作成）して保持する方法をとるものもある が、ネットワーク転送をできるだけ避けるために、 「得たいデータが失われていたら前のデータから再生成する」というアプローチをとっている。 そのために以下の制約?前提条件を持っている。 ? RDDはイミュータブルであること ? 元データのデータソースは信頼性が高く(高可用/高堅牢) 、イミュータブルである こと
- 10. 10Copyright ? 2013 NTT DATA Corporation 例えば、 のようにデータが生成された後に、 のように3個目～５個目のRDDの一部が失われたものとする。次に5個目のRDDを再取得すると、 のように2個目のRDDから必要なデータを再生成される。これは先の結果と同じ結果である。 再生成により同じ結果が得られるのは、RDDがイミュータブルだからである。 同様に1個目のRDDから後が失われていた場合は、信頼のおけるイミュータブルなデータソースから 再生成すればよい、ということになる RDD フォールトトレラント性を実現するための特性 RDD外部 RDD RDD RDD RDDRDD外部 RDD RDD RDD RDDRDD外部 RDD RDD RDD
- 11. 11Copyright ? 2013 NTT DATA Corporation RDDの依存関係 ? RDDの変換において、変換の元(親)のRDDと変換後 (子)のRDDの間には依存関係が定義される ? 依存関係は、スケジューリング(RDDの変換チェインを 処理可能なタスクに分割する)に関係する ? RDDの変換の依存関係には、以下の2種類がある ? 狭い依存関係（Narrow Dependency） ? 広い依存関係（Wide Dependency）
- 12. 12Copyright ? 2013 NTT DATA Corporation 狭い依存関係の例 RDD1 パーティション1-1 This is a pen. This is a book. What is this? パーティション1-2 Thank you. No Problem. This is a cup. RDD2 パーティション2-1 This is a pen. This is a book. パーティション2-2 This is a cup. 「This is」 Filterの例 親パーティションが単一の子パーティションの生成に関わっている依存関係
- 13. 13Copyright ? 2013 NTT DATA Corporation 広い依存関係の例 RDD1 パーティション1-1 dog, white dog, black cat, white パーティション1-2 cat, black dog, brown cat, brown RDD2 パーティション2-1 dog, white dog, black パーティション2-2 GroupByKeyの例 dog, brown cat, black cat, brown cat, white キーに基づく グルーピング キー?バリュー形 式のデータセット 親パーティションが、複数の子パーティションの生成に関わっている依存関係
- 14. 14Copyright ? 2013 NTT DATA Corporation ?主な登場人物 ? クラスタを構成するノード ? クライアント : ジョブのキックを担当 ? マスタ : クラスタ全体のリソース管理を担当 ? ワーカ : 計算資源の提供/管理を担当 ? ノード以外の主要な要素 ? ドライバ : ユーザがRDDの変換を記述したプログラム ? エグゼキュータ : ワーカ上で動作し、実際の計算処理を担当 ? タスク : ジョブを、エグゼキュータが処理可能な粒度に分解した処理単位 ? スケジューラ : ジョブをタスクに分割したり、タスクのエグゼキュータへの割り当てを 担当 RDDの変換が、クラスタ上でどのように動作するか
- 15. 15Copyright ? 2013 NTT DATA Corporation RDDの変換が、クラスタ上でどのように動作するか
- 16. 16Copyright ? 2013 NTT DATA Corporation ? エグゼキュータに割り当てる「タスク」はどのように生成され る？系譜との関係は？ ? タスクはどのようにエグゼキュータに割り当てられる？ ?系譜をもとにタスクを生成したり、エグゼキュータへの割り当 てをコントロールするのはスケジューラの役割 RDDの変換が、クラスタ上でどのように動作するか
- 17. 17Copyright ? 2013 NTT DATA Corporation ? 系譜をステージに分割する ? ステージの実行要否を判定する ? タスクを生成する ? タスクを実行する場所を決定する ? タスクの実行順序をスケジューリングする タスクの生成、実行までの流れ
- 18. 18Copyright ? 2013 NTT DATA Corporation ? DAGSchedulerが、系譜をステージに分割する ? ステージは系譜中で狭い依存関係が連続して発生する範囲である(依 存関係の種類は変換の種類によって決まる) 系譜をステージに分割する RDD ステージ
- 19. 19Copyright ? 2013 NTT DATA Corporation ステージ分割のステップ 起点を移しながらRDD間の 依存関係の解決とステー ジの成長、及びステージ間 の依存関係を解決し、系 譜がステージに分割される 起点から狭い依存関係でた どれるRDDに起点を移し、親 RDDを再帰的にたどって依 存関係の解決とステージの 成長を繰り返す。 狭い依存関係で閉じた範囲 が限定された段階で当該ス テージが確定する RDDの変換の方向 狭い依存関係 広い依存関係 RDD 成長中のステージ 確定したステージ ステージ間の依存関係 新規にステージを生成し、 起点のRDDを加える 変換の最後に出現する RDD(Actionが適用され るRDD)を、系譜をたどる 起点とする 起点から系譜をたどり、 親ノードとの依存関係 を解決する。 狭い依存関係を持つ RDDを現在のステージ に含め、ステージを成 長させる 現在の起点から広い依存 関係でたどれるRDDに起点 を移し、同様に親RDDとの 依存関係の解決とステージ の成長を繰り返す。 新なステージは広い依存関 係の子のRDDが属するス テージの親ステージとなる 変換の最後に出現 するRDD
- 20. 20Copyright ? 2013 NTT DATA Corporation ? パーティションごとにエグゼキュータ1つがまとめて計算できる 変換の範囲を決めるため ? 広い依存関係を発生させる変換は、パーティション内の要素 を子RDDのパーティションを処理する各エグゼキュータに分配 する必要があるので、エグゼキュータ間の通信が発生する。 なぜステージに分割するのか？
- 21. 21Copyright ? 2013 NTT DATA Corporation 広い依存関係を発生させる変換の実行例 4個 3個 1個 3個 countByKey countByKey関数で、図形ごと、色ごとに数をカウントする 2個 2個 1個 RDD 変換関数 パーティション RDD内の要素 グループ RDD1 RDD2 パーティショナによって要素の振り分け方が異なる。 この例の場合、色ごとに要素を振り分ける
- 22. 22Copyright ? 2013 NTT DATA Corporation 広い依存関係を発生させる変換の実行例
- 23. 23Copyright ? 2013 NTT DATA Corporation ? Sparkでは同一スケジューラで制御される複数のジョブで、RDDを共有することができ る ? 共有しているRDDがすでに計算済みで、メモリやディスクに実体を持つ場合、当該 RDDを生成するための前段のステージの実行を省略することができる。 ? 明示的にRDDをキャッシュした場合 ? ステージ内の最後のRDD(シャッフル直前のRDD) ステージの実行要否を判定する 他のジョブで キャシュされた RDD 後続のRDDがすでに計 算済みなので、親ス テージの実行は不要 親ステージ 子ステージ
- 24. 24Copyright ? 2013 NTT DATA Corporation ? DAGSchedulerが、実行対象の個々のステージに対してタスクを定義する ? 各ステージにおいて、ステージ内の最後のRDDのパーティション数から、当該ステージ のタスク数が決まる ? ステージに含まれるRDDの変換チェインから、タスクあたりの処理範囲が決まる タスクの生成 filter (青い図形を除外) map (図形の形を変換) union パーティション数 = 4なので、こ のステージのタスク数は4。 個々のタスクが異なるエグゼ キュータで実行される ステージ内の最後のRDDのパー ティションから、親RDDの方向に 一意にたどれる。 一意にたどれる範囲が1タスク。
- 25. 25Copyright ? 2013 NTT DATA Corporation ? RDDにはプリファードロケーションが定義されている場合がある。DAGSchedulerは、タ スクを実行するエグゼキュータを選ぶにあたって、プリファードロケーションをヒントに用 いる。 ? RDDにプリファードロケーションが定義されていない場合は、親RDDを通り掛け順にた どり、最初に見つかったプリファードロケーションを用いる。 タスクの実行場所の決定
- 26. 26Copyright ? 2013 NTT DATA Corporation ? プリファードロケーションはRDDの種類ごとに定義される。HDFSをデー タソースとするRDDはプリファードロケーション = ブロックの実体が配置 されているノードなので、データローカリティが担保される。 ? データソースをもとに生成されるRDDはプリファードロケーションを持つ ので、いつかはプリファードロケーションが見つかる。 タスクの実行場所の決定
- 27. 27Copyright ? 2013 NTT DATA Corporation ? ステージを構成するタスク群は「タスクセット」としてTaskSchedulerに渡される。 ? TaskSchedulerはタスクセット単位で実行順序のスケジューリングを行う。 ? TaskSchedulerはタスクセットの実行順序を決定するまでにタスクセットを登録してお く「プール」を1つ以上保持する。スケジューリング方式によってプールの数や使い方が 異なる。 ? 標準で2つのスケジューリング方式が用意されている ? FIFO ? 単一のプールを用いて、キューに様に扱う。プールに登録されたタスクセットを順にスケジューリング する ? FAIR ? 複数のプールを用いて、各プールから公平にタスクセットを取り出し、スケジューリングする。 タスクの実行順序のスケジューリング
- 28. 28Copyright ? 2013 NTT DATA Corporation ? Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing （http://www.cs.berkeley.edu/~matei/papers/2012/nsdi_spark.pdf） ? Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing （http://www.cs.berkeley.edu/~matei/papers/2011/tr_spark.pdf） ? Spark: Cluster Computing with Working Sets （http://www.cs.berkeley.edu/~matei/papers/2010/hotcloud_spark.pdf） （参考）論文紹介