Apache Spark：内部には何がありますか？

エントリー

最近、Apache Sparkプロジェクトは大きな注目を集めており、多数の小さな実用的な記事が書かれ、Hadoop 2.0の一部になりました。 さらに、彼はSpark Streaming、SparkML、Spark SQL、GraphXなどの追加フレームワークですぐに大きくなり、これらの「公式」フレームワークに加えて、さまざまなコネクタ、アルゴリズム、ライブラリなど、多くのプロジェクトが登場しました。 Sparkには他のBerkeleyプロジェクト（BlinkDBなど）のあらゆる種類の基本的な要素が含まれているという事実を考慮すると、深刻なドキュメントがないため、この動物園をすばやく自信を持って理解できます。これは簡単な作業ではありません。 したがって、私は忙しい人々の生活を楽にするためにこの記事を書くことにしました。

少しの背景：

Sparkは、2009年頃に始まったUC Berkeleyのラボプロジェクトです。 Sparkの創設者はデータベース分野の有名な科学者であり、彼らの哲学によれば、Sparkは何らかの形でMapReduceに対する答えです。 Sparkは現在、Apacheの「屋根」の下にありますが、イデオロギー家とコア開発者は同じ人たちです。

ネタバレ：2ワードのスパーク

Sparkは、次のような1つの文で説明できます。これは、大規模な並列DBMSのエンジンの内部です。 つまり、Sparkはストレージを昇格させませんが、他のものよりも優先されます（HDFS-分散ファイルシステムHadoopファイルシステム、HBase、JDBC、Cassandraなど）。 真実は、IndexedRDDプロジェクトはすぐに言及する価値があるということです-Sparkのキー/値ストレージは、おそらくすぐにプロジェクトに統合されるでしょう。また、Sparkはトランザクションを気にしませんが、そうでなければMPP DBMSエンジンです。

RDD-Sparkのコアコンセプト

Sparkを理解する鍵はRDD：Resilient Distributed Datasetです。 実際、これは信頼できる分散テーブルです（実際、RDDには任意のコレクションが含まれていますが、リレーショナルテーブルのようにタプルを操作するのが最も便利です）。 RDDは完全に仮想であり、その発生方法を把握するだけで、たとえばノードに障害が発生した場合に復旧できます。 そして、それを具体化することができます-分散、メモリまたはディスク（またはディスクへの押し出しを伴うメモリ）。 また、内部的に、RDDはパーティション化されています。これは、各作業ノードで処理されるRDDの最小量です。







Sparkで発生する興味深いことはすべて、RDDの操作を通じて発生します。 つまり、通常、Sparkのアプリケーションは次のようになります-RDDを作成し（たとえば、HDFSからデータを取得します）、それを台無しにし（map、reduce、join、groupBy、aggregate、reduce、...）、結果で何かを行います-たとえば、 HDFS。



さて、すでにこの理解に基づいて、Sparkは、タスクを調整するマスターと、実行に参加する多数の作業ノードが存在する、複雑な分析タスクの並列環境と見なすべきです。



このような単純なアプリケーションを詳細に見てみましょう（Scalaで作成します-これはこのファッショナブルな言語を学ぶ機会です）。

Sparkアプリケーションの例（すべてを含むわけではありません、インクルードなど）

各ステップで何が起こるかを個別に分析します。

def main(args: Array[String]){ // ,    val conf = new SparkConf().setAppName(appName).setMaster(master) val sc = new SparkContext(conf) //    HDFS,   RDD val myRDD = sc.textFile("hdfs://mydata.txt") //      .    . //      ,    (   // ) -  ""  val afterSplitRDD = myRDD.map( x => ( x.split(" ")( 0 ), x ) ) //    :  -    val groupByRDD = afterSplitRDD.groupByKey( x=>x._1 ) //  -     val resultRDD = groupByRDD.map( x => ( x._1, x._2.length )) //       HDFS resultRDD.saveAsTextFile("hdfs://myoutput.txt") }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

そこで何が起こっていますか？

では、このプログラムを調べて、何が起こるか見てみましょう。



まず、プログラムはクラスターのマスターで実行され、データの並列処理が行われる前に、1つのスレッドで静かに何かをする機会があります。 さらに-既に目立っているように-RDDでの各操作は異なるRDDを作成します（saveAsTextFileを除く）。 同時に、ファイルに書き込むか、たとえばマスターにアップロードするように要求した場合にのみ、RDDはすべて遅延して作成されます。実行が開始されます。 つまり、コンベアによってクエリプランのように実行されます。コンベア要素はパーティションです。



HDFSファイルから作成した最初のRDDはどうなりますか？ SparkはHadoopとうまく統合されているため、各作業ノードで独自のデータサブセットがアップロードされ、パーティション（HDFSの場合はブロックと一致）によってアップロードされます。 つまり、すべてのノードが最初のブロックをダウンロードし、計画に従って実行がさらに進みました。



ディスクから読み取った後、マップがあります-各作業ノードで簡単に実行されます。



次はgroupByです。 これはもはや単純なパイプライン操作ではなく、実際の分散グループ化です。 幸いなことに、この演算子はあまり賢く実装されていませんが、データの局所性の追跡が不十分であり、分散ソートに匹敵するパフォーマンスになるため、この演算子を避ける方が良いでしょう。 さて、これは考慮すべき情報です。



groupByの実行時の状況について考えてみましょう。 すべてのRDDは以前にパイプライン化されていました。つまり、どこにも何も保存されませんでした。 障害が発生した場合、彼らは再び不足しているデータをHDFSから取得し、パイプラインに渡します。 しかし、groupByはパイプラインを壊し、その結果、キャッシュされたRDDを取得します。 損失が発生した場合、すべてのRDDをgroupByに完全にやり直す必要があります。



Sparkの複雑なアプリケーションの障害によりパイプライン全体を再計算する必要がある状況を回避するために、Sparkではユーザーがpersistステートメントでキャッシュを制御できるようにします。 メモリ（この場合、メモリでデータが失われると再カウントが発生します-キャッシュがオーバーフローすると発生する可能性があります）、ディスク（常に十分に高速ではない）、またはキャッシュオーバーフローの場合はディスクへの排出を伴うメモリにキャッシュできます。



その後、再びマップとHDFSのエントリがあります。



さて、Sparkの内部で何が起こっているかは、単純なレベルで多かれ少なかれ明らかです。

しかし、詳細はどうですか？

たとえば、groupBy操作の仕組みを知りたいです。 または、reduceByKey操作、およびそれがgroupByよりもはるかに効率的である理由。 または、joinとleftOuterJoinの仕組み。 残念ながら、これまでの詳細のほとんどは、Sparkのソースからのみ、またはメーリングリストで質問することで最も簡単に学ぶことができます（ちなみに、Sparkで深刻なまたは非標準の操作を行う場合は、購読することをお勧めします）。



さらに悪いことに、さまざまなSparkコネクタで何が起こっているのかを理解しています。 そして、それらをどれだけ使用できるか。 たとえば、Sparkコネクタのサポートが理解できないため、Cassandraとの統合という考えを一時的に放棄しなければなりませんでした。 しかし、近い将来、高品質のドキュメントが登場することを期待しています。

Sparkの上にどんな面白いものがありますか？

• SparkSQL：Spark上のSQLエンジン。 すでに見たように、Sparkeには、ストレージ、インデックス、および独自の統計を除いて、これに関するほとんどすべての機能が既に備わっています。 これは最適化を非常に複雑にしますが、SparkSQLチームは新しい最適化フレームワークを見ていると主張しており、AMP LAB（Sparkが育った研究所）はSharkプロジェクトを拒否しません-Apache HIVEの完全な代替品
• Spark MLib：これは本質的にApache Mahaoutの代替であり、はるかに深刻です。 効率的な並列機械学習（RDDだけでなく、追加のプリミティブも使用）に加えて、SparkMLは、Breezeネイティブ線形代数パッケージを使用してローカルデータを処理し、Fortranコードをクラスターに引き付けます。 まあ、非常によく設計されたAPI。 簡単な例：相互検証クラスターで同時にトレーニングします。
  
  
• BlinkDB：​​非常に興味深いプロジェクト-大量のデータに加えて不正確なSQLクエリ。 一部のフィールドの平均を計算したいのですが、5秒以内に（精度を失って）それを実行したいです-お願いします。 与えられたもの以下のエラーを持つ結果が必要です-それも適切です。 ところで、このBlinkDBの一部はSpark内で見つけることができます（これは別のクエストと見なすことができます）。
  
  
• さて、多くの多くのことがSparkの上に書かれているので、私は自分の観点から最も興味深いプロジェクトのみをリストしました