pysparkでタイトル類似度を計測する #Pyspark

概要

大量のタイトルの類似度を測って、似ているタイトルのアイテムを列挙したい。
その場合、タイトルの類似度を図るために、(N*N)/2個のタイトルを比較しなければならない。

タイトルが増えていくと、一つのCPUだと辛くなるので、分散処理環境で並列に処理しようとおもい、pysparkで分散環境で出来ないか調査してみた。

コードはgistにあげている。

spark 2.2.1
hadoop 2.7

試したCSV

結果

cid_1	title_1	cid_2	title_2	simularity
2	いいい	6	あいう	0.33333333333333337
2	いいい	10	あいあいいいああうあいあ	0.25
2	いいい	11	あいあいうあいあうあいいあうあい	0.1875
2	いいい	4	ああい	0.33333333333333337
2	いいい	8	ああいああ	0.19999999999999996
2	いいい	5	いいう	0.6666666666666667
2	いいい	9	あいあいあいあいあいあい	0.25
6	あいう	2	いいい	0.33333333333333337

使ったシェル

df1 = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").load(sample.csv.gz)

csvを読み込む。圧縮してても良い。

df1 = df1.repartition(分割するサイズ).cache()
test1 = df1.take(1)

読み込んだデータフレームを分割する。本当は自動で分割してれるそうなのだが、この処理がないと1つのCPUしか使わなかったのでこの処理を入れた。本当に必要なのかよくわからない。

この登録を参考にした。

タイトルの類似度はリーベルシュタイン距離を測ることで調べた

filter(1.0 - (func.levenshtein(df1.title_1,df2.title_2) / func.greatest(func.length(df1.title_1),func.length(df2.title_2))) > 0.75)\

1.0 - levenshtein(タイトル1,タイトル2) / max(タイトル1,タイトル2) で、0.0-1.0までの類似度が求められ、0.0が全く似ていない。1.0が一致するということらしい。

pysparkで２つの変数にmaxを取るのに、max(a,b)というものが使えなかった。その代わり、greatest(a,b)を使って、maxを取った。

dfout.write.save('sampledir.',compression='snappy')

出力結果は snappy.parquet形式で出される。

dfout.explain()

結果の表示

dfout.show(100)

ただし、日本語で表示する際には、例外が発生した。
pysparkでutf-8を出すためには、環境変数を指定する。

export PYTHONIOENCODING=utf8

amazon EMRを利用してpysparkの動かした。

11万件なので、組み合わせの数は、N*(N-1)になり、約121億件の組み合わせでタイトル類似度を測った。集計時間は、m3.2xlarge　5台で、7時間程度、金額にして35ドル程度かかった。

マシンを、5台借りて7時間かかるのは流石に辛いので、別のアプローチを取ってみた。
類似のタイトルを探せれば良いので、simstringを使って、類似タイトルを検索してみた。

上と同じデータセット11万件を使い、GCPのn2-standardで10分程度で集計が出来た。
どのくらいの制度が出ているかは今から確認するが、sparkだとお金がかかりすぎるので、こちらのアプローチをもう少しやってみたい。