問題文の意味がわからない人のための 言語処理100本ノック 第9章 ベクトル空間法 (I)

0 はじめに

言語処理100本ノックは、東北大学の乾・岡崎研究室で公開されている、実践的な課題に取り組みながら，プログラミング，データ分析，研究のスキルを楽しく習得することを目指した問題集です。
これまでに、「第4章 形態素解析」、「第5章 係り受け解析」、「第8章 機械学習」を解いてきました。
引き続き「第9章 ベクトル空間法 (I)」を進めていきます。

0.1 この章でやること

enwiki-20150112-400-r10-105752.txt.bz2は，2015年1月12日時点の英語のWikipedia記事のうち，約400語以上で構成される記事の中から，ランダムに1/10サンプリングした105,752記事のテキストをbzip2形式で圧縮したものである．このテキストをコーパスとして，単語の意味を表すベクトル（分散表現）を学習したい．第9章の前半では，コーパスから作成した単語文脈共起行列に主成分分析を適用し，単語ベクトルを学習する過程を，いくつかの処理に分けて実装する．第9章の後半では，学習で得られた単語ベクトル（300次元）を用い，単語の類似度計算やアナロジー（類推）を行う．

要するにword2vecと似たような動きをするアルゴリズムを実装します。
word2vecっていうのは、単語をベクトルで表現して、単語同士の計算や比較をできるようにするアルゴリズムで、例えば「イチロー」-「野球」+「本田」=「サッカー」とかやるやつですね。
word2vecを聞いたことがない方はこの辺の記事(自然言語処理に新風を巻き起こしたWord2Vecとは何か)などを読んでおくとイメージがつくと思います。
word2vecには、既にGoogleやDL4Jなどの実装がありますが、この章ではこれらと似たような動きをするかもしれないアルゴリズムを自分で実装して、word2vecが何をやっているのかイメージをつかみます。
ちなみに、次の10章では、Googleのword2vecを使ってゴニョゴニョやるようです。

0.2 具体的には、下記のことを行います

1. Wikipediaの記事を読みこんで、単語文脈行列を作成する。 (80〜84)
2. 単語文脈行列の次元を圧縮して、単語ベクトルを作成する。(85)
3. 作成した単語ベクトルで遊んでみる。(86〜89)

0.3 ハマりどころ

この章までたどり着いた人なら、問題文の意味さえ理解できれば、コードに落とし込んでいくこと自体はそれほど難しくないと思います。
ただし実行できません。
データ量が大きすぎるんです。
問題文には

なお，問題83を素直に実装すると，大量（約7GB）の主記憶が必要になる． メモリが不足する場合は，処理を工夫するか，1/100サンプリングのコーパスenwiki-20150112-400-r100-10576.txt.bz2を用いよ．

とあります。
こう書かれると処理の工夫でなんとかしたくなりますよね。
私の場合は、84までは処理の工夫でなんとかなったんですが、100本ノック最大の難関という噂もある「85. 主成分分析による次元圧縮」に関しては、試行錯誤してみたものの歯が立たず、特異値分解で代替しました。
特異値分解で、かつ、1/100コーパスでも85を実行するには7時間ぐらいかかりました。
マシンのスペックとどれだけ効率のよいロジックを考えられるか次第ではありますが、まずは「1/100サンプリングのコーパス」で最後まで通して実装してみることをお勧めします。
その後、余裕があれば1/10サンプリングデータに挑戦してみると良いでしょう。

0.4 実行環境/ツール

ハードウェアとしては、CPU: 2.8GHz Intel Core i7 / メモリ: 16GB のMacBookProを、
ソフトウェアとしては、Sparkを使いました。(Apache Sparkとは？：Hadoopに続く分散処理のフレームワーク)
最初はSparkを使わずに普通にPythonで進めていたのですが、「83. 単語／文脈の頻度の計測」辺りで、ファイルに実行結果を保存するだけで2時間以上かかるようになってしまったので、あれこれ試行錯誤したあげく、Sparkを使うに至りました。Sparkなら数分で終わります。Sparkのサンプルコードの多くがScalaで書かれていることから、実装言語もScalaに切り替えました。(SparkにはPython向けのAPIもあります。)

0.5 全体構成

Treeコマンド

$ tree ./src
./src
└── main
    ├── resources
    │   ├── combined_words.txt
    │   ├── context_matrix
    │   ├── enwiki-20150112-400-r10-105752.txt
    │   ├── enwiki-20150112-400-r100-10576.txt
    │   └── word_vectors
    └── scala
        └── nlp100_9
            ├── Main.scala
            └── WikiRDD.scala

Main.scalaのmainメソッドの中から、

• runStage1() Wikipediaの記事を読みこんで、単語文脈行列を作成する。 (80〜84)
• runStage2() 単語文脈行列の次元を圧縮して、単語ベクトルを作成する。(85)
• runStage3() 作成した単語ベクトルで遊んでみる。(86〜89)

を順に呼び出して行きます。

Main.scala

package nlp100_9

import org.apache.spark._
import nlp100_9.WikiRDD._

object Main {
  val RAW_FILE = "/Users/inaba/Dropbox/NLP100-Spark/src/main/resources/enwiki-20150112-400-r100-10576.txt"
  val COMBINED_WORDS = "/Users/inaba/Dropbox/NLP100-Spark/src/main/resources/combined_words.txt"
  val CONTEXT_MATRIX = "/Users/inaba/Dropbox/NLP100-Spark/src/main/resources/context_matrix"
  val WORD_VECTORS = "/Users/inaba/Dropbox/NLP100-Spark/src/main/resources/word_vectors"
  val sc = new SparkContext(new SparkConf().setAppName("NLP100").setMaster("local[*]"))

  def main(args: Array[String]): Unit = {
    runStage1()
    runStage2()
    runStage3()
  }

コードを都度もれなく掲載すると読みにくそうなので、説明の途中では、その問題を解に直接関係する断片のみ記載し、コードの全体は最後にまとめて記載しています。

1 Wikipediaの記事を読みこんで、単語文脈行列を作成する。 (80〜84)

メソッドrunStage1()から80〜84の実装を呼び出しています。

Main.scala

  /**
    * 1. Wikipediaの記事を読みこんで、単語文脈行列を作成する。 (80〜84)
    * (100分の1ファイルの場合、8分程度で完了)
    */
  def runStage1(): Unit = {
    sc.textFile(RAW_FILE)
      .cleansData                          // 80. コーパスの整形
      .replaceCombinedWord(COMBINED_WORDS) // 81. 複合語からなる国名への対処
      .contexts                            // 82. 文脈の抽出
      .map(_ -> 1L)                        // RDD[(String, String)]型をRDD[((String, String), Long)]型に変換
      .sumByTC                             // 83.単語／文脈の頻度の計測(単語tと文脈語cの共起回数(f(t,c))を算出)
      .contextMatrix                       // 83.単語／文脈の頻度の計測(f(t,c)以外)と、84.単語文脈行列の作成
      .saveContextMatrix(CONTEXT_MATRIX)   // 単語文脈行列をファイルに保存する
  }

時間がもったいないので、余計なファイル保存は行わずに、80〜84まで一気に実行しています。

実際の処理はWikiRDD.scalaに記述しています。

ちなみに、RDDっていうのはSparkが提供しているListみたいなもんです。裏で上手く遅延評価＋分散処理してくれるところがListとの主な違いです。(参考:RDDについて簡単にまとめてみた)

80. コーパスの整形

文を単語列に変換する最も単純な方法は，空白文字で単語に区切ることである． ただ，この方法では文末のピリオドや括弧などの記号が単語に含まれてしまう．
そこで，コーパスの各行のテキストを空白文字でトークンのリストに分割した後，各トークンに以下の処理を施し，単語から記号を除去せよ．

• トークンの先頭と末尾に出現する次の文字を削除: .,!?;:()[]'"
• 空文字列となったトークンは削除

以上の処理を適用した後，トークンをスペースで連結してファイルに保存せよ．

これは問題無いですね。問題文の指示以外にも幾つか気になったゴミがあったので、削除対象文字を追加しています。

WikiRDD.scala

class WikiRDDString(rdd: RDD[String]) extends Serializable {
  def cleansData: RDD[String] = {

    def remove_garbage_char(word: String): String = {
      val reg = ".,!?;:()\\[\\]\\'\\\"“”"
      word.replaceAll("^[%s]+|[%s]+$".format(reg, reg), "")
    }

    rdd.map(sentence => (sentence split ' ').map(word => remove_garbage_char(word)).filter(word => !word.isEmpty).mkString(" "))
  }

81. 複合語からなる国名への対処

英語では，複数の語の連接が意味を成すことがある．
例えば，アメリカ合衆国は"United States"，イギリスは"United Kingdom"と表現されるが，"United"や"States"，"Kingdom"という単語だけでは，指し示している概念・実体が曖昧である．
そこで，コーパス中に含まれる複合語を認識し，複合語を1語として扱うことで，複合語の意味を推定したい．
しかしながら，複合語を正確に認定するのは大変むずかしいので，ここでは複合語からなる国名を認定したい．

インターネット上から国名リストを各自で入手し，80のコーパス中に出現する複合語の国名に関して，スペースをアンダーバーに置換せよ．
例えば，"United States"は"United_States"，"Isle of Man"は"Isle_of_Man"になるはずである．

まだ怖い言葉は出てきてないですね。
http://www.listofcountriesoftheworld.com/ をコピーしてcombined_words.txtとして保存しておいて、下記を実行します。
国名だけじゃなく地名・人名なども気にはなりますが、あまりこだわらずに先に進んだほうが吉です。

WikiRDD.scala

class WikiRDDString(rdd: RDD[String]) extends Serializable {
  (中略)
  def replaceCombinedWord(fileName: String): RDD[String] = {
    val combinedWordMap = Source.fromFile(fileName).getLines
      .filter(_.split(' ').length > 1)
      .map(combinedWord => (combinedWord, combinedWord.replace(' ', '_')))
      .toMap

    def replaceCombinedWords(sentence: String): String = {
      var s = sentence
      combinedWordMap.foreach {
        combinedWord => s = s.replaceAll(combinedWord._1, combinedWord._2)
      }
      s
    }

    rdd.map(replaceCombinedWords)
  }

82. 文脈の抽出

81で作成したコーパス中に出現するすべての単語tに関して，単語tと文脈語cのペアをタブ区切り形式ですべて書き出せ．
ただし，文脈語の定義は次の通りとする．

• ある単語tの前後d単語を文脈語cとして抽出する（ただし，文脈語に単語tそのものは含まない）
• 単語tを選ぶ度に，文脈幅dは{1,2,3,4,5}の範囲でランダムに決める．

難しい言葉は出てこないものの、何のためにこんなことをするのか意図の理解に苦しむかもしれません。
ある単語がどういった文脈で使われることが多いのかを調べるというのがやりたいことです。
で、その文脈ってのをどういう風に判断するかに色々なやり方があるっぽいんですが、ここではランダムに前後1〜5語を取ってきて文脈としようと言っているわけです。

例えば、「Wikipedia is an Internet encyclopedia, supported and hosted by the non-profit Wikimedia Foundation.」という文章があったとしたら、この問題で得たいアウトプットは、下記の様なタブ区切りの文字列になります。（下記の例では文脈幅は順に、3, 1, 5 であると仮定しています）

82のアウトプットイメージ

Wikipedia is
Wikipedia an
Wikipedia Internet
is Wikipedia
is an
an Wikipedia
an is
an Internet
an encyclopedia
an supprted
an and
an hosted
(以下略)

WikiRDD.scala

class WikiRDDString(rdd: RDD[String]) extends Serializable {
  (中略)
  def contexts: RDD[(String, String)] = {
    def getContexts(sentence: String): List[(String, String)] = {
      val words = sentence.split(' ')
      words.zipWithIndex.map { case (word, index) =>
        val d = scala.util.Random.nextInt(5) + 1      // 文脈幅
        val from = List(0, index - d + 1).max         // 文脈幅が0より前にならないように注意!
        val to = List(index + d, sentence.length).min // 文脈幅が文章の末尾より後ろにならないように注意!
        val contexts = words.slice(from, index).toList ::: words.slice(index + 1, to).toList
        contexts.map(context => (word, context))
      }.toList.flatten
    }
    rdd.filter(_.nonEmpty).flatMap(getContexts)
  }

83. 単語／文脈の頻度の計測

82の出力を利用し，以下の出現分布，および定数を求めよ．

• f(t,c): 単語tと文脈語cの共起回数
• f(t,∗): 単語tの出現回数
• f(∗,c): 文脈語cの出現回数
• N: 単語と文脈語のペアの総出現回数

83でやりたいことをSQLでいうと下記の様になります。本当にSQLを使う訳ではありません。あくまでイメージです。

# 82の出力結果のDDL
CREATE TABLE word_context (
  word VARCHAR,
  context VARCHAR
);

# f(t,c): 単語tと文脈語cの共起回数
SELECT word, context, COUNT(*) FROM word_context GROUP BY word, context;

# f(t,∗): 単語tの出現回数
SELECT word, COUNT(*) FROM word_context GROUP BY word;

# f(∗,c): 文脈語cの出現回数
SELECT context, COUNT(*) FROM word_context GROUP BY context;

# N: 単語と文脈語のペアの総出現回数
SELECT COUNT(*) FROM word_context;

これをScalaで書くわけですが、「f(t,c): 単語tと文脈語cの共起回数」以外は84でやってしまったほうが効率が良いので、ここではf(t,c)だけ出します。

WikiRDD.scala

class WikiRDDLong(rdd: RDD[((String, String), Long)]) {
  def sumByTC: RDD[((String, String), Long)] = rdd.groupByKey.map { case (k, v) => k -> v.sum }

ちょっとここだけ見ると謎かも知れませんが、最後にコードの全体像を載せているので確認して下さい。

84. 単語文脈行列の作成

83の出力を利用し，単語文脈行列Xを作成せよ．ただし，行列Xの各要素Xtcは次のように定義する．
f(t,c)≥10ならば，Xtc = PPMI(t,c) = max { log(N × f(t,c) / f(t,∗) × f(∗,c)), 0}
f(t,c)<10ならば，Xtc = 0
ここで，PPMI(t,c)はPositive Pointwise Mutual Information（正の相互情報量）と呼ばれる統計量である．
なお，行列Xの行数・列数は数百万オーダとなり，行列のすべての要素を主記憶上に載せることは無理なので注意すること．
幸い，行列Xのほとんどの要素は0になるので，非0の要素だけを書き出せばよい．

単語と文脈語のペアの出現頻度を元に、単語と文脈語の関係の強さを、正の相互情報量という統計量で表します。一見、難しげですが、問題文に書いている定義通りにコード化していくだけです。

WikiRDD.scala

class WikiRDDLong(rdd: RDD[((String, String), Long)]) {
  def contextMatrix: RDD[((String, String), Double)] = {
    // (83の残タスク) 単語tの出現回数: f(t,∗)
    val mapT = rdd.map(x => x._1._1 -> x._2).groupByKey.map { case (k, v) => k -> v.sum }.collect.toMap

    // (83の残タスク) 文脈語cの出現回数: f(∗,c)
    val mapC = rdd.map(x => x._1._2 -> x._2).groupByKey.map { case (k, v) => k -> v.sum }.collect.toMap

    // (83の残タスク) 単語と文脈語のペアの総出現回数
    val n = rdd.count

    // PPMIの算出
    // Xtc = PPMI(t,c) = max { log(N × f(t,c) / f(t,∗) × f(∗,c)), 0}
    rdd.filter(_._2 >= 10).map(x => x._1 -> Math.log(n * x._2 / (mapT(x._1._1) * mapC(x._1._2)))).filter(_._2 > 0)
  }

84のアウトプットイメージ

perceive they 4.3694478524670215
NFL teams 4.23410650459726
European elections 2.995732273553991
strictly is 1.791759469228055
Valley region 2.6390573296152584
head football 2.8903717578961645
Wilson was 0.6931471805599453
applied may 1.6094379124341003
population Census 2.302585092994046
private investment 4.543294782270004
(以下略)

2 単語文脈行列の次元を圧縮して、単語ベクトルを作成する。(85)

ここまでで「84のアウトプットイメージ」の様に、単語と文脈語との関係を定量化することができました。次はこれをベクトルとして表現していきます。

85. 主成分分析による次元圧縮

84で得られた単語文脈行列に対して，主成分分析を適用し，単語の意味ベクトルを300次元に圧縮せよ．

300次元！？

さてここで次元という言葉が出てくるわけですが、何なんでしょう300次元って？
「84で得られた単語文脈行列」を圧縮して300次元にするわけですから、現時点ではもっと大きい次元ということですね。

まず、「84で得られた単語文脈行列」を、単語を縦軸(行)、文脈を横軸(列)とする、スカスカの行列として捉えるんです。「84のアウトプットイメージ」の例だと下記の様な行列になります。

	they	teams	elections	is	region	football	was	may	Census	investment	(以下略)
perceive	4.36944	0	0	0	0	0	0	0	0	0
NFL	0	4.23410	0	0	0	0	0	0	0	0
European	0	0	2.99573	0	0	0	0	0	0	0
strictly	0	0	0	1.79175	0	0	0	0	0	0
Valley	0	0	0	0	2.63905	0	0	0	0	0
head	0	0	0	0	0	2.89037	0	0	0	0
Wilson	0	0	0	0	0	0	0.69314	0	0	0
applied	0	0	0	0	0	0	0	1.60943	0	0
population	0	0	0	0	0	0	0	0	2.30258	0
private	0	0	0	0	0	0	0	0	0	4.54329
(以下略)

この横幅を次元と呼んでる訳です。
1/10サンプリングデータだと、9.4万行ｘ10万列(10万次元)ぐらいになります。

最終的に、やりたいことは最初に述べたように、「イチロー」-「野球」+「本田」=「サッカー」の様な単語同士の比較・演算です。
こういった比較・演算を行うには10万次元もあっては無駄が大きいので、本当にその単語の性質を表現するのに必要な次元にまで圧縮するのです。
単語を表現するベクトルがあれば比較・演算はできるので、圧縮後には横軸(列)のラベル(上の例で言うところの、they, team, elections, ...)は失われてしまっても問題ありません。単語の性質を表す10万個の指標をまとめて、新しい300個の指標を作るイメージです。

圧縮する方法には、「主成分分析(PCA)」と「特異値分解(SVD)」があります。

主成分分析(PCA)

主成分分析の正しくて分かりやすい説明をする能力は、私にはないのでこのへんを参考にしてみてください。
ここでは、次元を削減するツールとしてブラックボックス的に理解して先に進めます。

主成分分析するコードは、Spark/Scalaでは下記の様になります。

    // 単語文脈行列のスカスカ行列化。PCAするためにはRowMatrix型にする必要がある。
    val contextMap = rdd.collect.toMap
    val words = rdd.keys.map(_._1).distinct
    val word_count = words.count.toInt
    val contexts = rdd.keys.map(_._2).distinct.collect

    val contextMatrix: RDD[Vector] = words.map(w => Vectors.dense(contexts.map(c => contextMap.getOrElse((w, c), 0.0)).slice(0, word_count)))

    val rowMatrix = new RowMatrix(contextMatrix)

    // PCA実行
    val pcMatrix = rowMatrix.computePrincipalComponents(300)

ただし、Sparkは65,535次元までしか対応していないらしく、下記のようなエラーが出ます。

Exception in thread "main" java.lang.IllegalArgumentException: Argument with more than 65535 cols: 100113

そこで、65,535次元以下になるように、適当にデータを絞って実行してみます。すると、今度はメモリ不足だと怒られます。

Exception in thread "main" java.lang.OutOfMemoryError: Java heap space

結局、主成分分析をする前に、5千次元ぐらいまで絞らないとcomputePrincipalComponentsを実行できませんでした。主成分分析する前にそんな大胆に絞ったら、主成分分析で次元削減する意味がないので、別の手を考える必要があります。

特異値分解(SVD)

次元を削減するためのもう一つのツールが特異値分解です。正確な説明が気になる方は、ググってみてください。
ここでもブラックボックスとして、大きな行列に対して特異値分解(computeSVD)を実行すると、UとΣとVという３つの行列に分解されて、UとΣの内積を計算すると次元圧縮ができるというふうに理解して先に勧めます。

Spark/Scalaでのコードは下記のようになります。

WikiRDD.scala

class WikiRDDDouble(rdd: RDD[((String, String), Double)]) {

  def wordVectors: RDD[(String, Array[Double])] = {
    // 単語文脈行列をマップ化
    val contextMap: Map[(String, String), Double] = rdd.collect.toMap

    // 一意な単語cのRDD
    val words: RDD[String] = rdd.keys.map(_._1).distinct

    // 一意な文脈語cの配列
    val contexts: Array[String] = rdd.keys.map(_._2).distinct.collect

    // 単語文脈行列を疎行列化
    val contextMatrix: RDD[Vector] = words.map(w => Vectors.dense(contexts.map(c => contextMap.getOrElse((w, c), 0.0))))

    // 特異値分解(SVD)
    val svd = new RowMatrix(contextMatrix).computeSVD(300, computeU = true)

    // 特異値分解後のUとΣの内積
    val us: RowMatrix = multiplyMatrix(svd.U, svd.s)

    words.zip(us.rows.map(_.toArray))
  }

  /**
    * 特異値分解後のUとΣの内積を算出する(+正規化)
    * 
    * https://github.com/sryza/aas/blob/master/ch06-lsa/src/main/scala/com/cloudera/datascience/lsa/RunLSA.scala#L131
    * と
    * https://github.com/sryza/aas/blob/master/ch06-lsa/src/main/scala/com/cloudera/datascience/lsa/RunLSA.scala#L143
    * を混ぜて実装
    */
  def multiplyMatrix(mat: RowMatrix, diag: Vector): RowMatrix = {
    val sArr: Array[Double] = diag.toArray
    new RowMatrix(mat.rows.map(vec => {
      val newArr: Array[Double] = (0 until vec.size).toArray.map(i => vec.toArray(i) * sArr(i))
      val length = math.sqrt(newArr.map(x => x * x).sum)
      Vectors.dense(newArr.map(_ / length))
    }))
  }

特異値分解なら、1/100データで7時間でなんとか実行することができました。
問題文の指示通りではないので、そのうちリベンジしたいものですが、参考文献に挙げたドキュメントを見ても大抵特異値分解を使っているようでしたので、一旦このまま進めても問題は無いでしょう。

3 作成した単語ベクトルで遊んでみる。(86〜89)

お疲れ様でした。事実上、ここまでで9章はほぼ終わりです。
後は、85で作った単語ベクトルをあれこれいじってみて性能を確認するだけです。

Main.scala

  def runStage3(): Unit = {
    // 単語ベクトルをファイルから復元
    val words = sc.textFile(WORD_VECTORS).restoreWordVectors

    // コサイン類似度を算出
    def cos(x: Array[Double], y: Array[Double]): Double = x.zip(y).map(xy => xy._1 * xy._2).sum

    def vec(word: String): Array[Double] = words.vector(word).get

    def plusVec(vec1: Array[Double], vec2: Array[Double]) = vec1.zip(vec2).map(v => v._1 + v._2)

    def minusVec(vec1: Array[Double], vec2: Array[Double]) = vec1.zip(vec2).map(v => v._1 - v._2)

    def printTopN(vec: Array[Double], n: Int = 10): Unit = {
      val wordMap = words.collect.toMap

      words
        .map(x => x._1 -> cos(vec, wordMap(x._1))).collect.toList // コサイン類似度の一覧を作成してリスト化
        .sortBy(_._2).reverse.slice(0, n) // ソートしてトップNを抽出
        .foreach(x => println(x._1 + "\t" + "%.8f".format(x._2))) // 表示
    }

    /**
      * 86. 単語ベクトルの表示
      *
      * 85で得た単語の意味ベクトルを読み込み，"United States"のベクトルを表示せよ．
      * ただし，"United States"は内部的には"United_States"と表現されていることに注意せよ．
      */
    println("86. 単語ベクトルの表示")
    println(vec("United_States").mkString(" "))

    /**
      * 87. 単語の類似度
      *
      * 85で得た単語の意味ベクトルを読み込み，"United States"と"U.S."のコサイン類似度を計算せよ．
      * ただし，"U.S."は内部的に"U.S"と表現されていることに注意せよ．
      */
    println("87. 単語の類似度")
    println(cos(vec("United_States"), vec("U.S")))
    println(cos(vec("United_Kingdom"), vec("U.K")))
    println(cos(vec("England"), vec("U.K")))
    println(cos(vec("United_Kingdom"), vec("England")))

    /**
      * 88. 類似度の高い単語10件
      *
      * 85で得た単語の意味ベクトルを読み込み，"England"とコサイン類似度が高い10語と，その類似度を出力せよ．
      */
    println("88. 類似度の高い単語10件")
    printTopN(vec("England"), 20)
    printTopN(vec("Obama"), 20)
    printTopN(vec("Japan"), 20)

    /**
      * 89. 加法構成性によるアナロジー
      *
      * 85で得た単語の意味ベクトルを読み込み，vec("Spain") - vec("Madrid") + vec("Athens")を計算し，そのベクトルと類似度の高い10語とその類似度を出力せよ．
      */
    println("89. 加法構成性によるアナロジー")
    printTopN(plusVec(minusVec(vec("Spain"), vec("Madrid")), vec("Athens")), 20)
    printTopN(plusVec(minusVec(vec("England"), vec("London")), vec("Tokyo")), 20)
  }

86. 単語ベクトルの表示

単語ベクトルを出力してみます。

vec("United_States")の出力結果

0.05194608599987417 0.0027944452235603725 -0.03464696786905431 0.09209495035824586 ...(以下略)

上記の様な、数字が300個表示されます。
単語文脈行列とは違い次元圧縮後のベクトルは、それだけを見ても意味はなく、他の単語との比較においてしか意味を持ちません。

87. 単語の類似度

United StatesとU.S.は0.8562344315089965と、高い類似度を示しましたが、それ以外はそれほど高くありませんでした。

88. 類似度の高い単語10件

問題文の指示以外にも幾つかの単語を確認してみました。ベスト20まで見ています。下記はprintTopN(vec("Obama"), 20)の結果ですが、米国の大統領を中心に政治家の名前やPresidentなどの単語が並んでいます。上手く学習できてるようですね。

88.類似度の高い単語10件

Obama   1.00000000
Obama's 0.98762456
Woodrow 0.97713770
Reagan  0.97351578
Barack  0.96941201
Nixon   0.91532997
Chen    0.90677497
Ronald  0.90399366
Carter  0.88981108
Jimmy   0.87959134
Abraham 0.82863200
Roosevelt   0.82592299
Clinton 0.82518346
Vice    0.82254346
Truman  0.81658883
Lyndon  0.80891514
Ulysses 0.80697392
Theodore    0.80415276
President   0.73341562
Harry   0.72970068

89. 加法構成性によるアナロジー

地理的な言葉であることは学習しているものの、国と首都の関係性までは理解できてないみたいですね。下記はprintTopN(plusVec(minusVec(vec("Spain"), vec("Madrid")), vec("Athens")), 20)の結果です。

89.加法構成性によるアナロジーの出力結果

Jawaharlal  0.81945976
McGill  0.81939535
Yeshiva 0.81939535
Emory   0.81939535
Tulane  0.81939535
Tezpur  0.81939535
Brandeis    0.81939535
McMaster    0.81939535
Washburn    0.81939535
Notre   0.81895540
Pontifical  0.80467503
Wesleyan    0.80453523
Freiburg    0.79883529
Loyola  0.79827056
Wuhan   0.79819758
EMUNI   0.79768679
Purdue  0.79070632
A&M 0.78805933
Middlesex   0.78464837
Wisconsin–Madison 0.78101616

ソースコード

Main.scala

package nlp100_9

import org.apache.spark._
import nlp100_9.WikiRDD._

object Main {
  val RAW_FILE = "/Users/inaba/Dropbox/NLP100-Spark/src/main/resources/enwiki-20150112-400-r100-10576.txt"
  val COMBINED_WORDS = "/Users/inaba/Dropbox/NLP100-Spark/src/main/resources/combined_words.txt"
  val CONTEXT_MATRIX = "/Users/inaba/Dropbox/NLP100-Spark/src/main/resources/context_matrix"
  val WORD_VECTORS = "/Users/inaba/Dropbox/NLP100-Spark/src/main/resources/word_vectors"
  val sc = new SparkContext(new SparkConf().setAppName("NLP100").setMaster("local[*]"))

  def main(args: Array[String]): Unit = {
    runStage1()
    runStage2()
    runStage3()
  }

  /**
    * 1. Wikipediaの記事を読みこんで、単語文脈行列を作成する。 (81〜84)
    * (100分の1ファイルの場合、8分程度で完了)
    */
  def runStage1(): Unit = {
    sc.textFile(RAW_FILE)
      .cleansData // 80. コーパスの整形
      .replaceCombinedWord(COMBINED_WORDS) // 81. 複合語からなる国名への対処
      .contexts // 82. 文脈の抽出
      .map(_ -> 1L) // RDD[(String, String)]型をRDD[((String, String), Long)]型に変換
      .sumByTC // 83.単語／文脈の頻度の計測(単語tと文脈語cの共起回数(f(t,c))を算出)
      .contextMatrix // 83.単語／文脈の頻度の計測(f(t,c)以外)と、84.単語文脈行列の作成
      .saveContextMatrix(CONTEXT_MATRIX) // 単語文脈行列をファイルに保存する
  }

  /**
    * 2. 単語文脈行列の次元を圧縮して、単語ベクトルを作成する。(85)
    * (100分の1ファイルをさらに1/100すると、分で完了)
    * (100分の1ファイルをさらに1/10すると、37分で完了)
    * (100分の1ファイルをそのまま突っ込むと、7時間13分で完了(00:23〜7:36))
    */
  def runStage2(): Unit = {
    sc.textFile(CONTEXT_MATRIX) // 単語文脈行列をファイルから復元
      .restoreContextMatrix // RDD[String]型をRDD[((String, String), Double)]型に変換
      .wordVectors // 次元を削減して単語ベクトル化する
      .saveWordVectors(WORD_VECTORS) // 単語ベクトルをファイルに保存する
  }

  /**
    * 3. 作成した単語ベクトルで遊んでみる。(86〜89)
    */
  def runStage3(): Unit = {
    val words = sc.textFile(WORD_VECTORS).restoreWordVectors

    def cos(x: Array[Double], y: Array[Double]): Double = x.zip(y).map(xy => xy._1 * xy._2).sum

    def vec(word: String): Array[Double] = words.vector(word).get

    def plusVec(vec1: Array[Double], vec2: Array[Double]) = vec1.zip(vec2).map(v => v._1 + v._2)

    def minusVec(vec1: Array[Double], vec2: Array[Double]) = vec1.zip(vec2).map(v => v._1 - v._2)

    def printTopN(vec: Array[Double], n: Int = 10): Unit = {
      val wordMap = words.collect.toMap

      words
        .map(x => x._1 -> cos(vec, wordMap(x._1))).collect.toList // コサイン類似度の一覧を作成してリスト化
        .sortBy(_._2).reverse.slice(0, n) // ソートしてトップNを抽出
        .foreach(x => println(x._1 + "\t" + "%.8f".format(x._2))) // 表示
    }

    /**
      * 86. 単語ベクトルの表示
      *
      * 85で得た単語の意味ベクトルを読み込み，"United States"のベクトルを表示せよ．
      * ただし，"United States"は内部的には"United_States"と表現されていることに注意せよ．
      */
    println("86. 単語ベクトルの表示")
    println(vec("United_States").mkString(" "))

    /**
      * 87. 単語の類似度
      *
      * 85で得た単語の意味ベクトルを読み込み，"United States"と"U.S."のコサイン類似度を計算せよ．
      * ただし，"U.S."は内部的に"U.S"と表現されていることに注意せよ．
      */
    println("87. 単語の類似度")
    println(cos(vec("United_States"), vec("U.S")))
    println(cos(vec("United_Kingdom"), vec("U.K")))
    println(cos(vec("England"), vec("U.K")))
    println(cos(vec("United_Kingdom"), vec("England")))

    /**
      * 88. 類似度の高い単語10件
      *
      * 85で得た単語の意味ベクトルを読み込み，"England"とコサイン類似度が高い10語と，その類似度を出力せよ．
      */
    println("88. 類似度の高い単語10件")
    printTopN(vec("England"), 20)
    printTopN(vec("Obama"), 20)
    printTopN(vec("Japan"), 20)

    /**
      * 89. 加法構成性によるアナロジー
      *
      * 85で得た単語の意味ベクトルを読み込み，vec("Spain") - vec("Madrid") + vec("Athens")を計算し，そのベクトルと類似度の高い10語とその類似度を出力せよ．
      */
    println("89. 加法構成性によるアナロジー")
    printTopN(plusVec(minusVec(vec("Spain"), vec("Madrid")), vec("Athens")), 20)
    printTopN(plusVec(minusVec(vec("England"), vec("London")), vec("Tokyo")), 20)
  }
}

WikiRDD.scala

package nlp100_9

import org.apache.spark.mllib.linalg.distributed.RowMatrix
import org.apache.spark.mllib.linalg.{Vector, Vectors}
import org.apache.spark.rdd.RDD

import scala.io.Source

object WikiRDD {
  implicit def addWikiRDDString(rdd: RDD[String]): WikiRDDString = new WikiRDDString(rdd)

  implicit def addWikiRDDLong(rdd: RDD[((String, String), Long)]): WikiRDDLong = new WikiRDDLong(rdd)

  implicit def addWikiRDDDouble(rdd: RDD[((String, String), Double)]): WikiRDDDouble = new WikiRDDDouble(rdd)

  implicit def addWikiRDDArray(rdd: RDD[(String, Array[Double])]): WikiRDDArray = new WikiRDDArray(rdd)
}

@SerialVersionUID(1L)
class WikiRDDString(rdd: RDD[String]) extends Serializable {

  /**
    * 80. コーパスの整形
    *
    * 文を単語列に変換する最も単純な方法は，空白文字で単語に区切ることである． ただ，この方法では文末のピリオドや括弧などの記号が単語に含まれてしまう．
    * そこで，コーパスの各行のテキストを空白文字でトークンのリストに分割した後，各トークンに以下の処理を施し，単語から記号を除去せよ．
    * - トークンの先頭と末尾に出現する次の文字を削除: .,!?;:()[]'"
    * - 空文字列となったトークンは削除
    *
    * 以上の処理を適用した後，トークンをスペースで連結してファイルに保存せよ．
    */
  def cleansData: RDD[String] = {

    // 問題文の指示以外にも気になったゴミを追加しています
    def remove_garbage_char(word: String): String = {
      val reg = ".,!?;:()\\[\\]\\'\\\"“”"
      word.replaceAll("^[%s]+|[%s]+$".format(reg, reg), "")
    }

    rdd.map(sentence => (sentence split ' ').map(word => remove_garbage_char(word)).filter(word => !word.isEmpty).mkString(" "))
  }

  /**
    * 81. 複合語からなる国名への対処
    *
    * 英語では，複数の語の連接が意味を成すことがある．
    * 例えば，アメリカ合衆国は"United States"，イギリスは"United Kingdom"と表現されるが，"United"や"States"，"Kingdom"という単語だけでは，指し示している概念・実体が曖昧である．
    * そこで，コーパス中に含まれる複合語を認識し，複合語を1語として扱うことで，複合語の意味を推定したい．
    * しかしながら，複合語を正確に認定するのは大変むずかしいので，ここでは複合語からなる国名を認定したい．
    *
    * インターネット上から国名リストを各自で入手し，80のコーパス中に出現する複合語の国名に関して，スペースをアンダーバーに置換せよ．
    * 例えば，"United States"は"United_States"，"Isle of Man"は"Isle_of_Man"になるはずである．
    */
  def replaceCombinedWord(fileName: String): RDD[String] = {
    val combinedWordMap = Source.fromFile(fileName).getLines
      .filter(_.split(' ').length > 1)
      .map(combinedWord => (combinedWord, combinedWord.replace(' ', '_')))
      .toMap

    def replaceCombinedWords(sentence: String): String = {
      var s = sentence
      combinedWordMap.foreach {
        combinedWord => s = s.replaceAll(combinedWord._1, combinedWord._2)
      }
      s
    }

    rdd.map(replaceCombinedWords)
  }

  /**
    * 82. 文脈の抽出
    *
    * 81で作成したコーパス中に出現するすべての単語tに関して，単語tと文脈語cのペアをタブ区切り形式ですべて書き出せ．
    * ただし，文脈語の定義は次の通りとする．
    * - ある単語tの前後d単語を文脈語cとして抽出する（ただし，文脈語に単語tそのものは含まない）
    * - 単語tを選ぶ度に，文脈幅dは{1,2,3,4,5}の範囲でランダムに決める．
    */
  def contexts: RDD[(String, String)] = {
    def getContexts(sentence: String): List[(String, String)] = {
      val words = sentence.split(' ')
      words.zipWithIndex.map { case (word, index) =>
        val d = scala.util.Random.nextInt(5) + 1
        val from = List(0, index - d + 1).max
        val to = List(index + d, sentence.length).min
        val contexts = words.slice(from, index).toList ::: words.slice(index + 1, to).toList
        contexts.map(context => (word, context))
      }.toList.flatten
    }
    rdd.filter(_.nonEmpty).flatMap(getContexts)
  }

  def restoreContextMatrix: RDD[((String, String), Double)] = rdd.map(_.split(' ') match {
    case Array(t: String, c: String, v: String) => t -> c -> v.toDouble
  })

  def restoreWordVectors: RDD[(String, Array[Double])] = rdd.map { x => val a = x.split(' '); a.head -> a.tail.map(_.toDouble) }
}

class WikiRDDLong(rdd: RDD[((String, String), Long)]) {
  /**
    * 83. 単語／文脈の頻度の計測
    *
    * 82の出力を利用し，以下の出現分布，および定数を求めよ．
    *
    * - f(t,c): 単語tと文脈語cの共起回数
    * - f(t,∗): 単語tの出現回数
    * - f(∗,c): 文脈語cの出現回数
    * - N: 単語と文脈語のペアの総出現回数
    */
  //    // メモリ不足の場合は単語tの開始文字毎にシーケンシャルに処理を進めると良いでしょう
  //    ('a' to 'z').foreach { c =>
  //      val tempRDD = rdd.filter(x => x._1.startsWith(c.toString) || x._1.startsWith(c.toString.toUpperCase))
  //      ...
  //    }
  //  }
  def sumByTC: RDD[((String, String), Long)] = rdd.groupByKey.map { case (k, v) => k -> v.sum }

  /**
    * 84. 単語文脈行列の作成
    *
    * 83の出力を利用し，単語文脈行列Xを作成せよ．ただし，行列Xの各要素Xtcは次のように定義する．
    * f(t,c)≥10ならば，Xtc=PPMI(t,c)=max{log(N×f(t,c) / f(t,∗)×f(∗,c)), 0}
    * f(t,c)<10ならば，Xtc=0
    * ここで，PPMI(t,c)はPositive Pointwise Mutual Information（正の相互情報量）と呼ばれる統計量である．
    * なお，行列Xの行数・列数は数百万オーダとなり，行列のすべての要素を主記憶上に載せることは無理なので注意すること．
    * 幸い，行列Xのほとんどの要素は0になるので，非0の要素だけを書き出せばよい．
    */
  def contextMatrix: RDD[((String, String), Double)] = {
    val rddTC = rdd.filter(_._2 >= 10)

    // 83. 単語tの出現回数: f(t,∗)
    val mapT = rdd.map(x => x._1._1 -> x._2).groupByKey.map { case (k, v) => k -> v.sum }.collect.toMap

    // 83. 文脈語cの出現回数: f(∗,c)
    val mapC = rdd.map(x => x._1._2 -> x._2).groupByKey.map { case (k, v) => k -> v.sum }.collect.toMap

    // 83. 単語と文脈語のペアの総出現回数
    val n = rdd.count

    // PPMIの算出
    // Xtc = PPMI(t,c) = max { log(N × f(t,c) / f(t,∗) × f(∗,c)), 0}
    rddTC.map(x => x._1 -> Math.log(n * x._2 / (mapT(x._1._1) * mapC(x._1._2)))).filter(_._2 > 0)
  }
}

class WikiRDDDouble(rdd: RDD[((String, String), Double)]) {

  def saveContextMatrix(fileName: String): Unit = {
    rdd.map { case ((t: String, c: String), v: Double) => Array(t, c, v).mkString(" ") }.saveAsTextFile(fileName)
  }

  /**
    * 85. 主成分分析による次元圧縮
    *
    * 84で得られた単語文脈行列に対して，主成分分析を適用し，単語の意味ベクトルを300次元に圧縮せよ．
    */
  def wordVectors: RDD[(String, Array[Double])] = {
    // 単語文脈行列をマップ化
    val contextMap: Map[(String, String), Double] = rdd.collect.toMap

    // 一意な単語cのRDD
    val words: RDD[String] = rdd.keys.map(_._1).distinct

    // 一意な文脈語cの配列
    val contexts: Array[String] = rdd.keys.map(_._2).distinct.collect

    // 単語文脈行列を疎行列化
    val contextMatrix: RDD[Vector] = words.map(w => Vectors.dense(contexts.map(c => contextMap.getOrElse((w, c), 0.0))))

    // 特異値分解(SVD)
    val svd = new RowMatrix(contextMatrix).computeSVD(300, computeU = true)

    // 特異値分解後のUとΣの内積
    val us: RowMatrix = multiplyMatrix(svd.U, svd.s)

    words.zip(us.rows.map(_.toArray))
  }

  /**
    * 特異値分解後のUとΣの内積を算出する
    * https://github.com/sryza/aas/blob/master/ch06-lsa/src/main/scala/com/cloudera/datascience/lsa/RunLSA.scala#L131
    * https://github.com/sryza/aas/blob/master/ch06-lsa/src/main/scala/com/cloudera/datascience/lsa/RunLSA.scala#L143
    * からコピー
    */
  def multiplyMatrix(mat: RowMatrix, diag: Vector): RowMatrix = {
    val sArr: Array[Double] = diag.toArray
    new RowMatrix(mat.rows.map(vec => {
      val newArr: Array[Double] = (0 until vec.size).toArray.map(i => vec.toArray(i) * sArr(i))
      val length = math.sqrt(newArr.map(x => x * x).sum)
      Vectors.dense(newArr.map(_ / length))
    }))
  }
}

class WikiRDDArray(rdd: RDD[(String, Array[Double])]) {
  def saveWordVectors(fileName: String): Unit = {
    rdd.map { case (t: String, v: Array[Double]) => List(t, v.mkString(" ")).mkString(" ") }.saveAsTextFile(fileName)
  }

  def vector(word: String): Option[Array[Double]] = {
    val wordVector = rdd.filter(_._1 == word)
    if (wordVector.isEmpty) {
      None
    } else {
      Option(wordVector.first._2)
    }
  }
}

参考文献

• 北野坂備忘録
• 言語処理100本ノック 第9章:ベクトル空間法 (I) @yamano357
• ディープラーニングチュートリアル 応用編
• Apache Spark入門 動かして学ぶ最新並列分散処理フレームワーク
• 岩波データサイエンス Vol.2 の「単語の意味をコンピュータに教える」
• Sparkによる実践データ解析 ―大規模データのための機械学習事例集 の「6章 潜在意味解析を使ったWikipediaの理解」
• 上記「6章 潜在意味解析を使ったWikipediaの理解」のコード

※ Spark / Scala / 自然言語処理 に詳しい方、変なところがあればツッコミを入れていただけると大変助かります。