分散データセットの依存性とキャッシュ SparkRDD

RDD 入門

RDD (Resilient Distributed Dataset) は分散データセットと呼ばれます。これは Spark における最も基本的なデータ抽象化です。これは、要素を並列に計算できる不変でパーティション化可能なコレクションを表します。 RDDはクラスです

RDD の特性

1. 各パーティションにアクセスするための優先場所を保存するリスト。 HDFS ファイルの場合、このリストには各パーティションが配置されているブロックの場所が格納されます。 「データの移動はコンピューティングの移動ほど良くない」という概念に従って、Spark はタスクをスケジュールするときに、処理するデータ ブロックの保存場所にコンピューティング タスクを割り当てるように最善を尽くします。

2. 各パーティションを計算する関数が保存されます。この計算方法は各データ ブロックに適用されます。 Spark での RDD の計算はシャードに基づいており、各 RDD はこの目標を達成するためにコンピューティング関数を実装します。計算関数は反復子を複合し、各計算の結果を保存する必要はありません。

3. RDD 間の依存関係。 RDD の各変換によって新しい RDD が生成されるため、RDD 間にパイプラインのような依存関係が形成されます。一部のパーティション データが失われた場合、Spark は RDD のすべてのパーティションを再計算する代わりに、この依存関係を通じて失われたパーティション データを再計算できます。

4. RDD パーティショニング関数 (Partitioner)。1 つはハッシュベースの HashPartitioner、もう 1 つは範囲ベースの RangePartitioner です。キー値 RDD の場合のみ、Partitioner が存在します。非キー値 RDD の場合、Partitioner の値は None です。 Partitioner 関数は、RDD 自体のシャード数を決定するだけでなく、親 RDD がシャッフルされるときのシャード数も決定します。

5. データ セットの基本コンポーネントであるパー​​ティションのグループ。 RDD の場合、各シャードはコンピューティング タスクによって処理され、並列コンピューティングの粒度を決定します。ユーザーは、RDD を作成するときに、RDD のシャードの数を指定できます。指定しない場合はデフォルト値が使用されます。デフォルト値は、プログラムに割り当てられた CPU コアの数です。

RDDの作成方法

1. コレクションをシリアル化してRDDを作成する（parallelize、makeRDD）

2. 外部データソース（testFile）を読み取る

3. 変換操作を通じて他のRDDを行RDDに変換する

RDD の 2 つの演算子:

1. 変革

• map(func) : func関数によって変換された各元の要素で構成される新しい分散データセットを返します。
• filter(func) : func関数が渡された後にtrueを返す元の要素で構成される新しいデータセットを返します。
• flatMap(func): map に似ていますが、各入力要素は 0 から複数の出力要素にマッピングされます (したがって、func 関数の戻り値は単一の要素ではなく Seq です)
• flatMap(func): map に似ていますが、各入力要素は 0 から複数の出力要素にマッピングされます (したがって、func 関数の戻り値は単一の要素ではなく Seq です)
• サンプル(置換あり、frac、シード):
• データは、分数で指定された割合に従ってサンプリングされます。置換に乱数を使用するかどうかを選択できます。 seed は乱数ジェネレータのシードを指定するために使用されます。
• union(otherDataset) : 元のデータセットとパラメータで構成される新しいデータセットを返します。
• ReduceByKey(func, [numTasks]) : (K, V) ペアのデータセットに適用され、指定された Reduce 関数を使用して同じキーを持つすべての値が集約された (K, V) ペアのデータセットを返します。 groupbykey と同様に、タスクの数は 2 番目のオプション パラメータを使用して設定できます。
• 結合(その他のデータセット、[タスク数]):
• タイプ (K,V) および (K,W) のデータセットで呼び出され、各キーのすべての要素をまとめた (K,(V,W)) ペアのデータセットを返します。
• groupWith(otherDataset, [numTasks]) : タイプ (K,V) および (K,W) のデータセットで呼び出され、要素が (K, Seq[V], Seq[W]) タプルであるデータセットを返します。この操作は、他のフレームワークでは CoGroup と呼ばれます。
• cartesian(otherDataset) : デカルト積。しかし、データセット T と U で呼び出されると、すべての要素が直交積で相互作用する (T、U) ペアのデータセットが返されます。
• 交差(otherDataset): ソースRDDとパラメータRDDの交差を検出した後、新しいRDDを返します。
• distinctive([numTasks])) ソースRDDの重複を排除した後、新しいRDDを返します。
• groupByKey([numTasks])は(K, V)のRDDで呼び出され、(K, Iterator[V])のRDDを返します。
• ReduceByKey(func, [numTasks]) (K, V) RDD で呼び出され、指定された Reduce 関数を使用して同じキーを持つ値を集約する (K, V) RDD を返します。 groupByKey と同様に、オプションの 2 番目の引数を使用して、reduce タスクの数を設定できます。
• aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])
• sortByKey([ascending], [numTasks]) は (K, V) RDD で呼び出されます。 K は Ordered インターフェースを実装する必要があります。キーでソートされた (K, V) RDD を返します。
• sortBy(func,[ascending], [numTasks]) sortByKeyに似ていますが、より柔軟です
• join(otherDataset, [numTasks]) は、タイプ (K,V) および (K,W) の RDD で呼び出され、同じキーを持つ要素のすべてのペアを含む (K,(V,W)) の RDD を返します。
• cogroup(otherDataset, [numTasks])は、(K,V)および(K,W)型のRDDで呼び出され、(K,(Iterable)を返します。

2.アクション

• Reduce(func) RDD のすべての要素を func 関数で集計します。この関数は可換かつ並列化可能である必要があります。
• collect() は、データセットのすべての要素をドライバー プログラム上の配列として返します。
• count()はRDD内の要素数を返します。
• first()はRDDの最初の要素を返します（take(1)と同様）
• take(n)はデータセットの最初のn要素からなる配列を返します。
• takeSample(withReplacement,num, [seed]) は、データセットからランダムにサンプリングされた num 要素で構成される配列を返します。不足部分を乱数で置き換えるかどうかを選択できます。 Seed は乱数ジェネレータのシードを指定するために使用されます。
• takeOrdered(n, [順序付け])
• saveAsTextFile(path) は、データセットの要素をテキストファイルとして HDFS ファイル システムまたはその他のサポートされているファイル システムに保存します。各要素に対して、Spark は toString メソッドを呼び出して、それをファイル内のテキストに変換します。
• saveAsSequenceFile(path) は、データセット内の要素を Hadoop シーケンスファイル形式で指定されたディレクトリに保存します。このディレクトリは、HDFS または Hadoop でサポートされている他のファイル システムである可能性があります。
• saveAsObjectFile(パス)
• countByKey() タイプ (K, V) の RDD の場合、各キーに対応する要素の数を示す (K, Int) マップを返します。
• foreach(func) は、更新するデータセットの各要素に対して関数 func を実行します。

RDDの依存関係

1. 狭い依存関係

狭い依存関係とは、各親 RDD のパーティションが最大 1 つの子 RDD のパーティションによって使用されることを意味します。

要約: 一人っ子のイメージメタファーへの狭い依存

2. 幅広い依存関係

広い依存関係とは、複数の子RDDパーティションが同じ親RDDパーティションに依存することを意味します。

要約: 超越のイメージメタファーへの狭い依存

3. 血統

RDD は、粗粒度の変換、つまり多数のレコードに対して実行される単一の操作のみをサポートします。失われたパーティションの回復を容易にするために、RDD を作成する一連の Lineage (系統) が記録されます。 RDD の系統は、RDD のメタデータ情報と変換動作を記録します。 RDD の一部のパーティション データが失われた場合、この情報に基づいて失われたデータ パーティションを再計算して復元できます。

DAG生成

DAG (Directed Acyclic Graph) は有向非巡回グラフと呼ばれます。元の RDD は、一連の変換を通じて DAG を形成します。 DAG は、RDD 間のさまざまな依存関係に応じて、さまざまなステージに分割されます。依存関係が狭い場合、パーティション変換プロセスはステージ内で完了します。幅広い依存関係の場合、シャッフルの存在により、親 RDD が処理された後にのみ次の計算を開始できます。したがって、幅広い依存関係がステージを分割するための基礎となります。

RDD キャッシュ

Spark が非常に高速である理由の 1 つは、異なる操作間でデータセットをメモリ内に保持またはキャッシュできることです。 RDD を永続化した後、各ノードは計算されたシャード結果をメモリに保存し、この RDD または派生 RDD の他のアクションで再利用します。これにより、後続のアクションが大幅に高速化されます。 RDD 関連の永続性とキャッシュは、Spark の最も重要な機能の 1 つです。キャッシュは、Spark が反復アルゴリズムと高速なインタラクティブ クエリを構築するための鍵であると言えます。

依存関係を見つけてステージを分割する目的の 1 つは、キャッシュを分割することです。ステージを分けてキャッシュを設定するには？

(1) 狭い依存関係でキャッシュを設定したい場合にキャッシュを使用する

（２）広範囲の依存関係にキャッシュを設定したい場合にはチェックポイントを使用する

キャッシュとチェックポイントを設定するにはどうすればいいですか?

キャッシュ: someRDD.cache() はキャッシュを正常に追加し、メモリに格納します。

someRDD.persist(StorageLevel.MEMORY_AND_DISK): 必要に応じてキャッシュの場所（メモリとハードディスク）を設定します

チェックポイント: RDDで計算されたデータをローカルディスクまたはHDFSに保存できます

sc.setCheckpointDIr("hdfs://hadoop1:9000/checkpoint") は、ワイド依存関係の前にチェックポイントパスを設定します。

someRDD.checkpoint() はチェックポイントを設定します

キャッシュとチェックポイントの違い

キャッシュはデータをキャッシュするだけで、RDD の依存関係は変更しません。チェックポイントは新しい RDD を生成し、後続の RDD は新しい RDD に依存するため、依存関係が変更されます。データ復旧の順序: チェックポイント---》キャッシュ--》再計算