Pysparkでコンテンツの類似度を図る

# はじめに
pysparkでコンテンツの類似度を図るためのコードを作成しました。
日本語の文献が少なかったので、メモします。
sparkを一から導入するのはめんどくさかったので、dockerを使用しました。

# dockerでpysparkの導入
## dockerのインストール
dockerの導入には
http://qiita.com/hshimo/items/e24b1fbfbf775ec7c941
を参考にしました。

基本的には
[Get started with Docker for Mac](https://docs.docker.com/docker-for-mac/)
よりdmgファイルをダウンロードしてきてインストールするだけです。

## spark+jupyterのdockerイメージの使用
spark+jupyterのdockerイメージをcloneします。
cloneしたのはhttps://github.com/busbud/jupyter-docker-stacks/tree/master/all-spark-notebookです。
README内に記されているように、

```bash
$ docker run -d -p 8888:8888 jupyter/all-spark-notebook -e GRANT_SUDO=yes
```
terminalで上のコマンドを打ちます。
これにより、pysparkを使用することができるjupyter notebookがlocalhost:8888に立ち上がります。
（-e GRANT_SUDO=yesはユーザー権限なしにjupyterを利用できるするようにするオプションです。）

docker psコマンドを打つと

```bash
$ docker ps
CONTAINER ID        IMAGE                        COMMAND                  CREATED             STATUS              PORTS                    NAMES
91fc42290759        jupyter/all-spark-notebook   "tini -- start-not..."   3 hours ago         Up 3 hours          0.0.0.0:8888->8888/tcp   nifty_bartik
```
というように、指定したdockerが起動していることがわかります。

# コード

作成したコードは
https://github.com/kenchin110100/machine_learning/blob/master/samplePysparkSimilarity.ipynb
にあります。

## 初期化
まず必要なライブラリをインポートして、pysparkの初期化を行います。

```python

# coding: utf-8
"""
pysparkで類似度を図るためのサンプル
"""
import numpy as np
from pyspark import SQLContext, sql
import pyspark
from pyspark.sql import functions, Row
from pyspark.mllib.linalg import DenseVector

sc = pyspark.SparkContext('local[*]')
sqlContext = sql.SQLContext(sc)
```
scはpysparkのRDD型を利用するために必要なインスタンス、
sqlContextはDataFrame型を利用するために必要なインスタンスです。

## サンプルデータの作成
次に類似度を図るためのサンプルデータを作成しました。

```python
# サンプルデータの作成
samples = [['aaa', 'a', 30, 1,2,3,4,5]  + np.random.randn(5).tolist(),
    ['aaa', 'b', 30,2,1,3,4,1] + np.random.randn(5).tolist(),
    ['bbb', 'a', 30,4,5,3,2,4] + np.random.randn(5).tolist(),
    ['bbb', 'b', 30,1,2,4,3,1] + np.random.randn(5).tolist(),
    ['ccc', 'a', 30,4,5,2,1,2] + np.random.randn(5).tolist(),
    ['ccc', 'b', 30,1,2,5,4,1] + np.random.randn(5).tolist(),]

# カラム名の作成
colnames = [
    'mc', 'mtc', 'area_cd',
    'label1', 'label2', 'label3', 'label4', 'label5',
    'label6', 'label7', 'label8', 'label9', 'label10'
]
colnames1 = [col + '_1' for col in colnames]
colnames2 = [col + '_2' for col in colnames]

# 作成したサンプルデータをpysparkのDataFrame型に変換
df1 = sqlContext.createDataFrame(sc.parallelize(samples), colnames1)
df2 = sqlContext.createDataFrame(sc.parallelize(samples), colnames2)
```
mc, mtcをユニークのkeyとして、label1~label10を素性として考えています。

同じサンプルデータを2つのデータフレームに格納しているのは、
類似度を図る組み合わせをjoinで作成するためのです。
samplesをsc.parallelize(samples)でRDD型に変換してから、createDataFrameでDataFrame型にしています。

## 組み合わせの列挙
次にDataFrame型のjoinを使用して、類似度を測定する組み合わせを列挙します。

```python
joined_df = df1.join(df2, df1.area_cd_1 == df2.area_cd_2).filter(functions.concat(df1.mc_1, df1.mtc_1) != functions.concat(df2.mc_2, df2.mtc_2))
```
DataFrame型のjoinは

```python
df1.join(df2, <条件>, 'left' or 'inner' or ...)
```
で作成することができます。
今回作成したコードでは、joinの後にfilterを行うことで、
自分自身との類似度を計らないようにしています。

```python
functions.concat(df1.mc_1, df1.mtc_1)
```
ではユニークにしたいmcとmtcの2つのkeyを結合させることで、1つのユニークなkeyとして扱っています。

## 類似度の計算
ここまでで作成したDataFrameを使用して、類似度の計算を行なっていきます。
### 関数の定義
まず関数を定義します。

```python
def match_sim(row1 ,row2):
    keys = row1.asDict().keys()
    total = len(keys)
    count = 0
    for key in keys:
        if row1[key] == row2[key]:
            count += 1
    return float(count)/total

def cosine_sim(vec1 ,vec2):
    dot = abs(vec1.dot(vec2))
    n1 = vec1.norm(None)
    n2 = vec1.norm(None)
    return float(dot/(n1*n2))
```
match_simはカテゴリ変数の類似度を測定するための関数で、
pysparkのRow型を渡します。
一致していれば1、一致していなければ0を返し、比較した素性数で割った値を返します。

cosine_simはコサイン類似度を計算するための関数で、
pyspark.mllibのDenseVector型を渡します。

この関数を利用して、行ごとの類似度を計算していきます。

### 類似度の計算

```python
joined_rdd = joined_df.rdd.map(lambda x: (
    Row(mc_1=x.mc_1, mtc_1=x.mtc_1, mc_2=x.mc_2, mtc_2=x.mtc_2),
    Row(label1=x.label1_1, label2=x.label2_1, label3=x.label3_1, label4=x.label4_1, label5=x.label5_1),
    DenseVector([x.label6_1,x.label7_1,x.label8_1,x.label9_1,x.label10_1]),
    Row(label1=x.label1_2, label2=x.label2_2, label3=x.label3_2, label4=x.label4_2, label5=x.label5_2),
    DenseVector([x.label6_2,x.label7_2,x.label8_2,x.label9_2,x.label10_2])
                                         )) \
.map(lambda x: (x[0], match_sim(x[1], x[3]), cosine_sim(x[2], x[4]))) \
.map(lambda x: (x[0].mc_1, x[0].mtc_1, x[0].mc_2, x[0].mtc_2, x[1], x[2]))

```
まずは、先ほど作成したDataFrame型であるjoined_dfをrdd型に変換してmapします（１行目）
5種類のデータ型を各組み合わせに対して保存します。
Row(mc_1=x.mc_1 ...)は類似度を図るユニークのkeyを保存するためのRow（2行目）
Row(label1=x.label1_1 ...)はカテゴリ変数を保存するためのRow（3行目）
DenseVector(x.label6_1, ...)は連続変数を保存するためのVector（4行目）
5行目、6行目は類似度を図るもう一方の行のカテゴリ変数、連続変数を保存します。

このように作成した５種類のデータ型を保存したRDDに対してさらにmapをします(8行目)。
x[0]はそのまま、x[1]とx[3]に対しては一致による類似度を、x[2], x[4]に対してはコサイン類似度を計算します。
最後に、再びDataFrame型に渡すために、整形します（9行目）。

### 結果の出力
このように作成した類似度テーブルは以下のようになります。

```python
sqlContext.createDataFrame(joined_rdd, ['tar_mc', 'tar_mtc', 'res_mc', 'res_mtc', 'match_sim', 'cosine_sim']).show()

+------+-------+------+-------+---------+--------------------+
|tar_mc|tar_mtc|res_mc|res_mtc|match_sim|          cosine_sim|
+------+-------+------+-------+---------+--------------------+
|   aaa|      a|   aaa|      b|      0.4|  0.2979433262317515|
|   aaa|      a|   bbb|      a|      0.2|  0.2161103600613806|
|   aaa|      a|   bbb|      b|      0.4|  0.6933162039799152|
|   aaa|      a|   ccc|      a|      0.0| 0.34941331375143353|
|   aaa|      a|   ccc|      b|      0.6|  0.5354750033557132|
|   aaa|      b|   aaa|      a|      0.4| 0.19428899651078324|
|   aaa|      b|   bbb|      a|      0.2| 0.10702152405150611|
|   aaa|      b|   bbb|      b|      0.2|  0.4033681950723296|
|   aaa|      b|   ccc|      a|      0.0| 0.20097172584128625|
|   aaa|      b|   ccc|      b|      0.4|  0.6861144738544892|
|   bbb|      a|   aaa|      a|      0.2|  0.3590385377694502|
|   bbb|      a|   aaa|      b|      0.2| 0.27266040008605663|
|   bbb|      a|   bbb|      b|      0.0|  1.1313716028957246|
|   bbb|      a|   ccc|      a|      0.4|0.009321106239696326|
|   bbb|      a|   ccc|      b|      0.0|  1.0017633803368193|
|   bbb|      b|   aaa|      a|      0.4|  0.2176828683879606|
|   bbb|      b|   aaa|      b|      0.2|   0.194213765887726|
|   bbb|      b|   bbb|      a|      0.0| 0.21381230488831227|
|   bbb|      b|   ccc|      a|      0.0| 0.21074015342537053|
|   bbb|      b|   ccc|      b|      0.6| 0.34536679942567616|
+------+-------+------+-------+---------+--------------------+
only showing top 20 rows
```
それぞれのkeyに対して、カテゴリの一致度による類似度とコサイン類似度が計算されています。


# 終わりに

今回はdockerのjupyter-sparkイメージを使用して、データ間の類似度を測定するためのサンプルを作成しました。
もっと簡潔な書き方もあるはずですが（MLlibのColumnSimilarityを使用するなど）、諸々の理由があり、今回はこのような回りくどいコードを書きました。
docker及びsparkの初心者なので、これからもう少し練習をしていきたいと思います。