一.原理

目的：根据某个word来查找包含该word的document。

现在，假设有一个输入文件input.data，里面包含5篇document，该文件的具体内容如下：

doc1 Apache Spark Scala Hadoop Java C Python Do And Will KNN

doc2 SVM Scala News Play Akka Yes GBDT

doc3 LDA SVM RF GBDT Adaboost Kmeans KNN

doc4 QQ BAT I Great All LDA

doc5 Apache Hadoop MapReduce Git SVN SVM

文件每行包含一篇doc的标示符(如doc1)以及该doc中包含的word，并且doc与word之间以”\t”分隔，而word与word之间以空格符分隔。

现在，我们需要建立的倒排索引如下(从结果数据中取出的top 10)：

(Akka,doc2)

(Python,doc1)

(QQ,doc4)

(RF,doc3)

(Apache,doc1|doc5)

(Will,doc1)

(Java,doc1)

(MapReduce,doc5)

(SVM,doc2|doc3|doc5)

(Scala,doc1|doc2)

(Git,doc5)

上述结果为k-v结构的pair对，key值为word，value为文档表示符，并且doc与doc之间以”|”分隔，以示区分。下面使用Spark来建立倒排索引。

二.实现

根据上面的分析，大概知道了该实例的目的，下面使用Scala编写Spark程序实现倒排索引。

/**

* Created by liudiwei on 2016-07-30.

*/

importorg.apache.spark.SparkContext

importorg.apache.spark.SparkConf

importorg.apache.spark.SparkContext._

importorg.apache.spark.SparkContext

importorg.apache.spark.rdd.RDD

importorg.apache.commons.configuration.{ PropertiesConfiguration => HierConf }

importorg.apache.spark.broadcast.Broadcast

object InvertedIndex{

def main(args : Array[String]){

val conf =newSparkConf().setAppName("invertedIndex")

.set("spark.serializer","org.apache.spark.serializer.JavaSerializer")

.set("spark.akka.frameSize","256")

.set("spark.ui.port","4071")

val sc =neworg.apache.spark.SparkContext(conf)

val cfg =newHierConf(args(0))

val inputfile = cfg.getString("inputfile")

val result = sc.textFile(inputfile)

.map(x => x.split("\t"))

.map(x => (x(0), x(1)))

.map(x => x._2.split(" ").map(y => (y, x._1)))

.flatMap(x => x)

.reduceByKey( (x, y) => x +"|"+ y)

result.collect.foreach(println)

sc.stop()

}

}

从上面的代码可以看出，在spark中，只需几个map操作再加上一个reduceByKey聚合函数就可以建立倒排索引，非常简洁，相比其他语言有很大的优势。整个代码的核心只有21-26行这一系列操作，可以说是非常精简。

三.参考文献

End.

作者：拾毅者

来源：『刘帝伟』维护的个人技术博客

本文均已和作者授权，如转载请与作者联系。