キャッシュと遅延評価
ここまで、実際にSparkSQLを利用したデータ操作を見てきました。前回に引き続き、ここではSparkの機能の一つであるRDDのキャッシュの概念を見ていきたいと思います。
RDDのキャッシュ
今回利用したコードではCSVの読み込みを行った後、データフレームのキャッシュを行っています。以下に当該部分のコードを抜粋します。
・・・（省略
val purchase = raw.filter( _ != head).map( _.split(",") ).map( x => Purchase(x(0).toInt, x(1).toInt, x(2).toInt)).toDF().cache
toDF関数を呼び出した後にcache関数の呼び出しを行っています。Sparkではこのようにcache関数を明示的に呼び出すことでデータをメモリ上にキャッシュし、高速な処理を実現します。
なぜキャッシュ処理が必要なのか
前回、Sparkの遅延評価について触れました。
Sparkはインメモリ処理を行えるため、高速な処理が可能と説明されますが、最初のデータの読み込み部分はディスクから読み出す必要がありますし、処理の機構上、すべてのデータをメモリ上に持ち続けることは難しいため、メモリ上で処理を効果的に行うための遅延実行の仕組みと、Sparkユーザーが明示的に再利用するデータをメモリ上に保持するキャッシュの仕組みが用意されています。
ここでは、SparkSQLの処理で利用したCSVファイルをキャッシュの仕組みが理解しやすいように、ただ画面に表示するだけの処理を作り、インタラクティブシェルを使って、遅延評価とキャッシュの仕組みを見ていきます。
以下のコードとSparkUI（http://localhost:4040/jobs）を利用してSparkの処理内容を見ていきます。
val raw = sc.textFile("./sample_purchase.csv")
raw.foreach(println) //1回目の画面表示
raw.foreach(println) //2回目の画面表示
以下はspark-shellを実行し、まだ何もコードを実行していない状態です。
この時のSparkUIは当然何のJobも実行されていない状態です。
次に以下のコードを実行します
val raw = sc.textFile("./sample_purchase.csv")
この時のSparkUIはどうでしょうか？
前回の記事で説明した通り、Sparkの処理はDAGと呼ばれる依存関係を積み上げ、実際に処理を行う必要ができた際に、DAGを遡り処理が実行されます。
実際にSparkに処理をさせてみます。以下のコードを実行します。
raw.foreach(println) //1回目の画面表示
画面にCSVの内容が表示されました。
SparkUIを確認してみます。
foreach(println)の処理は画面にデータを表示する必要があるので、依存関係を遡り処理が実行されています。
Stageの中身を見てみます。
Stageのページはさまざまな内容が表示されますが、赤枠で囲った部分を確認してみてください。赤枠で囲った部分は実際にデータがどのように読み込まれたか、どのくらいの量が読み込まれたかを表示する欄となります。なぜ2つに分けて読み込まれているか。パーティションの概念は別の回で紹介できればと考えているので、今回はそういうものだとして話を進めます。
今回のケースですと、データがファイルシステム（hadoop）から合計10件（4件と6件）読み込まれたことを表しています。
再度、foreach(println)で同じraw変数を表示してみます。以下のコードを実行します。
raw.foreach(println) //2回目の画面表示
SparkUIを確認します。
ちゃんと2回目の処理が実行されています。
Stageの方も確認してみます。
ここで注目していただきたいのは、同じraw変数への処理にもかかわらず、再度ファイルシステムからファイルの読み込みを行っている部分です。
Sparkでは遅延実行を行うため、同一データを複数の処理パターンで扱う場合も基本的にはデータの読み込み処理（ディスクアクセス）から実行されます。この挙動（毎回ファイルを読み込む）は、とても無駄が大きく、Sparkの処理を遅くする大きな原因になります。
この問題を解決するために、Sparkで実際のプログラムを構築する際には、RDDなどデータのキャッシュ機能をユーザーがある程度意識する必要があります。
次にSparkでファイルを読み込む際に、cache関数で実際にメモリ上にデータをストアさせてみましょう。以下のコードを実行します。
val raw = sc.textFile("./sample_purchase.csv").cache
Spark上でcache処理を行うのは非常に簡単で、キャッシュしたいデータに対しcache関数を呼び出すだけです。
補足
cache関数の他にデータをキャッシュする方法としてpersist関数が用意されています。cache関数はpersist関数にMEMORY_ONLYを渡した場合と同様の挙動となります。persist関数はcache関数より柔軟なキャッシュ方法を提供していますが、レプリケーションやシリアライズの考え方など必要となるのでここでは触れません。
次に再度foreach(println)を呼び出し、SparkUIからStageを確認します。
前回までと同様にデータがファイルシステムから読み込まれていますが、青枠部分の表示が以前までとは変わっているのが分かるでしょうか？ このようにキャッシュ処理を行った場合、SparkUI上でも表示方法が変わります。
この状態で再度foreach(println)を呼び出した場合はどうなるでしょうか？ foreach(println)を実行し、SparkUIを確認します。下記の図は2回目の処理の実行時のStageです。
図の赤枠に表示が「memory」と変更され、キャッシュからデータが読み込まれているのが見て取れます。
最後にStorageのタブを確認してみましょう。読み込んだ「sample_purchase.csv」ファイルが実際にメモリ上にキャッシュされているのが分かります。
まとめ
今回はSparkSQLを利用した簡単な分析について扱いました。SparkSQLを利用することで、map処理やreduce処理の組み合わせになれない方も普段使い慣れたSQLで解析処理を高速に行うこともメリットですし、今回プログラミングにフォーカスするということで取り扱いませんでしたが、Optimaizaに任せることで、RDDで処理を構築するよりも速度的なメリットを享受することも可能になります。この記事がSparkSQLの利用のきっかけになれば幸いです。
【PR】
