Spark MLib 每周一算法 —— Word2Vec

Word2Vec

Word2Vec是一个Estimator，顾名思义就是将给定的词(word)的序列转化成一个指定长度的向量值(vector)。Word2Vec的训练集就是一组语料，可以是任意文章的集合，通常将每个句子，即词的序列，作为一条输入项。Word2Vec的fit过程负责计算出这些语料中出现的词(word)的对应向量(vector)，从而生成Word2VecModel对象。transform过程就是根据每个单词的向量值求平均算出整个句子的向量值，从而给出最终结果。
实际在自然语言处理(nlp)的任务中，我们其实更关心的就是每个词的向量表达。我们可以把这个计算词向量表达的过程，即Word2Vec的整个fit过程，想象成——把一个词作为一个给定n维空间的点镶嵌到n维空间当中去，所以这个过程又被称为词嵌入(Embedding)。Embedding也可以认为是机器对人类语料进行学习的结果，机器从语料本身(不需要任何额外标注信息)得到了每个词的特征；获取了一种在机器世界表达词的方式，这种对词的表达方式可以用到后续任何语言学习任务当中去。

Embedding算法过程

Skip-Gram
Spark的Embedding采用的是Skip-Gram方法，即通过语料中给定的词，去预测该词周边出现的词。用户可以定义周边词的个数，这个个数是根据一个窗口阈值随机生成的。整个预测过程可以通过构建一个单隐层的神经网络来进行训练，输入和输出就是整个输入语料中符合条件的给定单词与周边词的配对(一般使用OneHotEncoder编码表示)，隐层的神经元个数就是词向量空间的维度数。最终需要的就是输入层到隐层之间的权重矩阵，该矩阵的每一个横向量就是对应词的词向量。

Hierarchical softmax
因为最终只需要输入层到隐层之间的权重矩阵，所以就不需要训练完整的神经网络的权重值，Spark采用hierarchical softmax的方式来简化训练过程，该方法不用计算出某个输入下所有输出层的神经元的softmax概率值，而是构建一个以词的出现频率为依据的二叉树结构，词频约高的结点越靠近二叉树的根，叶子结点为所有输出层神经元，即词典中所有的词。这样只需要计算$\log_{2}W$个结点的二分类概率值，然后找到代表某个预测词的叶子结点的路径就可，比计算所有词的数量W个结点的全部softmax概率值要高效很多，词量越大节省的效率越大。

参数

• inputCol: 需要进行转化词序列的属性名称，该属性列用一个数组来表达序列。 (没有默认值)
• outputCol: 转化后的结果向量属性名称。 (默认值: w2v_######__output)
• maxSentenceLength: 表示输入的句子(词序列)中最多包含多少个词，超过指定长度的序列将被截断成多个序列作为输入。(正整数，默认值: 1000)
• minCount: 表示输入词在输入语料中至少出现多少次，才会进行向量转化，少于该出现次数的次将会在输入值中直接丢弃。 (>=0整数，默认值: 5)
• vectorSize: 表示词向量空间的维度。 (正整数，默认值: 100)
• windowSize: 表示Skip-Gram方法中的词窗口阈值，这个阈值决定给定词的周边有多少个词会作为给定词的周边结果。比如真实窗口数为2，那就是前后各2个词作为给定词的预测结果，Spark在处理不同的给定词时会采用随机浮动的窗口。(正整数，默认值: 5)
• numPartitions: Spark的实现参数，在fit过程中将输入的句子(词序列)分成指定的n个protition处理，减少同一时间的内存占用，默认不分区。 (正整数，默认值: 1)

model对象的方法

• getVectors(): 获取所有单词对应的词向量，以DataFrame类型数据返回，包含word和vector两列。
• findSynonyms(word, num): 找出给定词的同义词，以DataFrame类型数据返回指定的最接近的num个同义词，包含word和similarity两列，其中word为近义词，similarity为近似程度，Spark采用给定词向量和所有词向量的夹角余弦相似度作为近似程度值。给定的词必须在模型对象的词向量列表中存在，不然会得到java.lang.IllegalStateException: xxx not in vocabulary异常
• findSynonymsArray(word, num): 和上面的findSynonyms(word, num)一样，找出给定词的同义词，但以Array[(String, Double)]类型的数据返回，每一条记录里包含了近义词和近似程度两个值。
• findSynonyms(vector, num): 找出给定向量的同义词，返回结果和上面的findSynonyms(word, num)一致，给定的向量可以是模型对象的词向量列表中不存在的任意符合维度的向量。
• findSynonymsArray(vector, num): 和上面的findSynonyms(vector, num)一样，找出给定词的同义词，返回值和findSynonymsArray(word, num)一样是一个Array[(String, Double)]数组。

例子

val wordDataFrame = spark.createDataFrame(Seq(
      (0, Array("Hi", "I", "heard", "about", "Spark")),
      (1, Array("I", "wish", "Java", "could", "use", "case", "classes")),
      (2, Array("Logistic", "regression", "models", "are", "neat"))
    )).toDF("id", "words")

    val word2Vec = new Word2Vec()
      .setInputCol("words")
      .setOutputCol("result")
      .setVectorSize(3)
      .setMinCount(0)
    val model = word2Vec.fit(wordDataFrame)

    model.getVectors.show(false)
    //    +----------+------------------------------------------------------------------+
    //    |word      |vector                                                            |
    //    +----------+------------------------------------------------------------------+
    //    |heard     |[-0.053989291191101074,0.14687322080135345,-0.0022512583527714014]|
    //    |are       |[-0.16293057799339294,-0.14514029026031494,0.1139335036277771]    |
    //    |neat      |[-0.0406828410923481,0.028049567714333534,-0.16289857029914856]   |
    //    |classes   |[-0.1490514725446701,-0.04974571615457535,0.03320947289466858]    |
    //    |I         |[-0.019095497205853462,-0.131216898560524,0.14303986728191376]    |
    //    |regression|[0.16541987657546997,0.06469681113958359,0.09233078360557556]     |
    //    |Logistic  |[0.036407098174095154,0.05800342187285423,-0.021965932101011276]  |
    //    |Spark     |[-0.1267719864845276,0.09859133511781693,-0.10378564894199371]    |
    //    |could     |[0.15352481603622437,0.06008218228816986,0.07726015895605087]     |
    //    |use       |[0.08318991959095001,0.002120430115610361,-0.07926633954048157]   |
    //    |Hi        |[-0.05663909390568733,0.009638422168791294,-0.033786069601774216] |
    //    |models    |[0.11912573128938675,0.1333899050951004,0.1441687047481537]       |
    //    |case      |[0.14080166816711426,0.08094961196184158,0.1596144139766693]      |
    //    |about     |[0.11579915136098862,0.10381520539522171,-0.06980287283658981]    |
    //    |Java      |[0.12235434353351593,-0.03189820423722267,-0.1423865109682083]    |
    //    |wish      |[0.14934538304805756,-0.11263544857501984,-0.03990427032113075]   |
    //    +----------+------------------------------------------------------------------+

    model.findSynonyms("I", 2).show(false)
    //    +-------+-------------------+
    //    |word   |similarity         |
    //    +-------+-------------------+
    //    |are    |0.800910234451294  |
    //    |classes|0.45088085532188416|
    //    +-------+-------------------+

    println("Synonyms Array: " +
      model.findSynonymsArray(Vectors.dense(0.1, 0.1, 0.1), 2).mkString(","))
    //  [(models,0.9970051050186157), (case,0.9669119715690613)]

    val testDataFrame = spark.createDataFrame(Seq(
      (0, Array("Hello", "I", "heard", "about", "Spark")),
      (1, Array("You", "know", "nothing")),
      (2, Array("You", "know", "Java"))
    )).toDF("id", "words")

    val result = model.transform(testDataFrame)
    result.show(false)
//    +---+-------------------------------+----------------------------------------------------------------+
//    |id |words                          |result                                                          |
//    +---+-------------------------------+----------------------------------------------------------------+
//    |0  |[Hello, I, heard, about, Spark]|[-0.0168115247040987,0.04361257255077362,-0.0065599825698882345]|
//    |1  |[You, know, nothing]           |[0.0,0.0,0.0]                                                   |
//    |2  |[You, know, Java]              |[0.04078478117783864,-0.01063273474574089,-0.0474621703227361]  |
//    +---+-------------------------------+----------------------------------------------------------------+