Spark mllibのPrefixSpan実装

# はじめに

この記事は[Apache Spark Advent Calendar 2015](http://qiita.com/advent-calendar/2015/apache-spark)の7日目の記事です。

Spark 1.5から系列パターンマイニングアルゴリズムの１つ PrefixSpanがmllibに実装されました。

我々は文書や検索クエリー内のキーワードの並びから有益な情報を抽出する目的でPrefixSpanアルゴリズムに以前より興味を持っていました。

OSSで公開されているPrefixSpan実装はオンメモリー上でデータを処理するため、大規模データを処理することが難しかったのですが、Sparkを使った分散PrefixSpan実装が登場したことで、大規模データに対して系列パターン抽出が可能になりました。

今回は、このSpark mllibの分散PrefixSpan実装が、どのくらいのサイズのデータを、どのくらいの処理時間で処理できるのかを確認するために、日本語ngramコーパス [2] 1000万件から頻出形態素パターンを抽出する処理を実行させた。

# 動作確認、ベンチマーク

## 確認方法

Spark上で分散処理可能なPrefixSpan実装がどのくらい大規模なデータを処理できるかを確認するために、日本語ngram web corpusから頻出形態素列を抽出する処理を実行させた。

使用したngram corpusは[2]で公開されているもので

- 7gram
- 1000万件

のデータをcassandraにinsertし、それをsparkからアクセスしてPrefixSpanで頻出形態素パターンを抽出することにした。

```
gm | partition | content                               | freq
----+-----------+---------------------------------------+------
  7 |         0 |           ( は ち おうじ みなみ の え |   18
  7 |         0 |            ) から 桓武 天皇 の 時代 ( |   26
  7 |         0 | ) について 】 当 キャンペーン 終了 後 |   15
  7 |         0 |                  ( まゆ ) の 間 を 、 |   11
  7 |         0 |          3 分 ブン の 2 以上 イジョウ |   11
  7 |         0 |    ( PDF ファイル ) ( ダウンロード し |   46
  7 |         0 |          ( 偶然 かも しれ ない けど ) |   38
  7 |         0 |              1 ) 予定 価格 調書 案 の |   20
  7 |         0 |                  1 号 店 3 階 の 販売 |   11
  7 |         0 |        22 日 、 東京 地裁 八王子 支部 |   32

```


## クラスター設定

使用したSparkクラスタ構成は以下である。

今回使用したクラスタは評価テスト用であり、処理時間性能の絶対性能を測定できるものではない。
あくまで、どの程度のデータを処理できるかを評価するためである。

- Spark master 1台, Spark worker 7台構成
- cassandra server 3台

マシンスペックは全台ほぼ同じで

- CPU: 4コア
- mem: 32GB

のマシンを使用した。

Sparkの設定は

- standalone mode
- worker memory: 4GB
- driver memory: 10GB

とした。
 
## 処理時間

PrefixSpanのパラメータ

- minSupportを 0.1〜 0.0001まで3段階に変化
- maxPatternLength = 3に固定

とした。

またデータはcassandraからRDDとして読み込んだ後cacheさせ、cassandraへのアクセスオーバーヘッドは最小限になるようにした。

処理時間は総処理時間をlinuxのtime commandで測定した。

| minsup | min count| time |
|:-----------|------------:|:------------:|
| 0.1       |   1,000,000 |     1m34.274s     |
| 0.01     |      100,000 |    4m26.014s   |
| 0.001       |       10,000 |     4m18.359s    |
| 0.0001      |         1,000|   30m35.717s     |


## 結果

minsupが小さくなればなるほど頻出itemの異なり数が大きくなり、それに伴い指数関数的に処理時間を大きくなることが予想され、実測値でもそうなった。1000万件の大規模データからcount >= 1000の頻出パターンをOOMなしに30分で抽出することができ、分散PrefixSpan実装の有効性を確認できた。



# Spark mllib PrefixSpan実装概説

mllib PrefixSpan実装
https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/mllib/fpm/PrefixSpan.scala

の処理フローを以下で概説します。

## 頻出itemの抽出とid化

PrefixSpan.rum methodで

- 頻出itemの抽出
- minCount以上の頻出itemのid化
- 系列パターンの内部表現化

が
https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/mllib/fpm/PrefixSpan.scala#L139
https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/mllib/fpm/PrefixSpan.scala#L154

あたりで行われています。

頻出itemの抽出はword countと同じで、結果はdriverにcollectされています。

## PrefixSpan本体処理 = getFreqPatterns
がhttps://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/mllib/fpm/PrefixSpan.scala#L179
https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/mllib/fpm/PrefixSpan.scala#L236

になります。

このmethod内で、分散PrefixSpanアルゴリズムが実装されています。

## itemによる射影

https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/mllib/fpm/PrefixSpan.scala#L256

が↑で実行されています。

https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/mllib/fpm/PrefixSpan.scala#L271
で
prefixとpostfixを2重のflatMapでjoinさせて頻出prefixとその頻度を求めています。
この結果はこれもdriverにcollectされています。

ここが、分散PrefixSpanアルゴリズムのコアのように思います。

この処理の中で射影dbのサイズがサイズが小さいものは、
https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/mllib/fpm/PrefixSpan.scala#L289
smallPrefixesとして保持されて、次のlocalPrefixSpan objectにより再帰的に処理されるようです。

## 各worker上でのlocalPrefixSpan.runの実行

https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/mllib/fpm/PrefixSpan.scala#L312

に実行されています。

## 最終的な頻出パターンの抽出結果

https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/mllib/fpm/PrefixSpan.scala#L317

でdriver上で保持されている頻出パターンと、最後に、各worker上のlocalPrefixSpan objectが抽出した
頻出パターンをRDDとしてunionして、返しています。

# おわりに

以前より興味を持っていた分散PrefixSpanのSpark mllib実装を1000件の大規模データで動作確認を実施した。
動作確認前は、実際に処理できる不安な面を持っていたが、minsupがある一定以上の場合は十分高速に処理できることを確認できた。

大規模コーパスや検索クエリーから、頻出キーワードパターン、形態素列パターンの抽出にPrefixSpanを適用する場合

1. 大規模データに対するPrefixSpan適用の困難
2. 無意味な頻出パターンの抽出
3. 情報が重複する、内包される頻出パターンの抽出

の3つの欠点があった。

1. はSpark mllib PrefixSpan実装で解決できそうである。2,3に関しては、処理させる大規模データへの前処理、後処理で解決することが多かったが、PrefixSpanの拡張アルゴリズムにより、2., 3. を解決したいと考えている。
例えば

- itemへの重要度ウェイトを付与して、そのウェイトにより抽出するパターンを制御する。
- PrefixSpanをClosed系列パターンマイニングにまで拡張する。

などである。これらをSpark mllib PrefixSpanアルゴリズムに適用できればと考えている。





# 参考情報

[1] PrefixSpan: http://chasen.org/~taku/software/prefixspan/
[2] 日本語ウェブコーパス 2010: http://s-yata.jp/corpus/nwc2010/