Sparkの永続化処理（後日談）

以前に「Sparkの永続化処理に参った話」という記事を書きましたが、その後わかったことなどを書きます。おすすめの設定なども書いていきます。

checkpoint()のおさらい

Sparkのデータの永続化手法には何種類かありますが、その中でもcheckpoint()の特徴としては、

• 即実行される。対抗馬であるpersist()は呼んでもマークされるだけでActionが実行されないと永続化処理が実行されないのと比べてタイミングが制御しやすい。
• 取り消せない。これもpersist()との比較になりますが、persist()にはunpersist()があるのに対して、checkpoint()にはそれがない。¹
• 遅い。ファイルに書き込むので仕方がない²ですが、途中経過を計算して一時的に保持しときたいなーなんていうライトな用途にはやや重い。persist()なら・・・。

取り消せないので永続化したファイルがずっと残ってしまってそれがファイル容量は食うし、遅いし、後片付けもしないといけないし面倒だ！というのが前回のお話でしたが、それがもう少しなんとかなったよ、という話です。

SparkのConfigurationを穴のあくほど見つめる

Spark公式のConfigurationを眺めていたところ、あるじゃないですか良い項目が！

プロパティ名	デフォルト	意味
spark.checkpoint.compress	false	RDDチェックポイントを圧縮するかどうか。 一般的には良い考えです。 圧縮にはspark.io.compression.codecが使用されます。
spark.cleaner.referenceTracking.cleanCheckpoints	false	参照がスコープ外の場合にチェックポイントファイルをクリーニングするかどうかを制御します。

共にデフォルトがfalseですが、チェックポイント使うならこれは必須でしょうよ！

試してみた

それなりに期待した通りに効きました。

まず、spark.checkpoint.compressですがこれは環境にもよりますが書き出し先がRAMディスクでもない限り³、効果あると思います。自分はSSDの劣化が怖くてHDDをチェックポイント出力先にしていたので効果はてきめんでした。

次にspark.cleaner.referenceTracking.cleanCheckpointsの方ですが、こちらは期待通りが半分、がっかりが半分でした。
期待通りなのは、当たり前ですがちゃんと消えてくれることで、自分はループに埋め込まれてて一回のセッションで数十回程度checkpoint()を呼んでるんですがファイルは多くても最新の2回分とかなのでディスク容量の圧迫はあまり心配しなくて良くなりました。
がっかりは、結局最後（セッションの終わり）に全部消えてくれるわけでもないので、別途お掃除は必要といったところです。
そもそも、どういった仕組みで参照が切れたと判断しているのかが不明なので、セッションを終わらせる前にひと手間かけるにしてもどうしたらいいのかわからない。
ファイルが消えるタイミングも「いらなくなったら即！」というわけでもないようなので、GCと同じ仕組みなのかな？

結論

どんな事情があるのかわかりませんが、個人的にはチェックポイントを使うなら

• spark.checkpoint.compress
• spark.cleaner.referenceTracking.cleanCheckpoints

の2つはほぼ必ずtrue！にしといたほうがいいと思います。

1. ただしpersist()はpersist()で、unpersist()忘れると（実質）メモリリークになるという使いずらさがありますが。malloc()/free()かと・・・。 ↩

2. 生ドライブでなくHDFSを使うともう少しましなのかもしれません。 ↩

3. つまりCPU時間を圧縮処理に使ってでもI/Oを減らした方がトータルで速くなる場合です。 ↩