Merge pull request #80 from zouzias/feature/moreLikeThis

Feature/more like this

Author: zouzias

project/build.properties

@@ -1 +1 @@
-sbt.version=0.13.12
+sbt.version=0.13.13

scripts/loadAlice.scala

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import scala.io.Source
+import org.zouzias.spark.lucenerdd._
+import org.zouzias.spark.lucenerdd.LuceneRDD
+val words = Source.fromFile("src/test/resources/alice.txt").getLines().map(_.toLowerCase).toSeq
+val rdd = sc.parallelize(words)
+val luceneRDD = LuceneRDD(rdd)
+luceneRDD.cache
+luceneRDD.count
+
+
+luceneRDD.moreLikeThis("_1", "alice adventures wonderland", 1, 1, 20).take(20).foreach(println)

src/main/scala/org/zouzias/spark/lucenerdd/LuceneRDD.scala

@@ -264,6 +264,22 @@ class LuceneRDD[T: ClassTag](protected val partitionsRDD: RDD[AbstractLuceneRDDP
     partitionsRDD.map(_.size).reduce(_ + _)
   }
 
+  /**
+    * Lucene's More Like This (MLT) functionality
+    * @param fieldName Field name
+    * @param query Query text
+    * @param minTermFreq Minimum term frequency
+    * @param minDocFreq Minimum document frequency
+    * @param topK Number of returned documents
+    * @return
+    */
+  def moreLikeThis(fieldName: String, query: String,
+                   minTermFreq: Int, minDocFreq: Int, topK: Int = DefaultTopK)
+  : LuceneRDDResponse = {
+    logInfo(s"MoreLikeThis field: ${fieldName}, query: ${query}")
+    partitionMapper(_.moreLikeThis(fieldName, query, minTermFreq, minDocFreq, topK), topK)
+  }
+
   /** RDD compute method. */
   override def compute(part: Partition, context: TaskContext): Iterator[T] = {
     firstParent[AbstractLuceneRDDPartition[T]].iterator(part, context).next.iterator

src/main/scala/org/zouzias/spark/lucenerdd/partition/AbstractLuceneRDDPartition.scala

@@ -118,6 +118,20 @@ private[lucenerdd] abstract class AbstractLuceneRDDPartition[T] extends Serializ
    */
   def phraseQuery(fieldName: String, query: String, topK: Int): LuceneRDDResponsePartition
 
+
+  /**
+    * Lucene's More Like This (MLT) functionality
+    * @param fieldName Field name
+    * @param query Query text
+    * @param minTermFreq Minimum term frequency
+    * @param minDocFreq Minimum document frequency
+    * @param topK Number of returned documents
+    * @return
+    */
+  def moreLikeThis(fieldName: String, query: String,
+                   minTermFreq: Int, minDocFreq: Int, topK: Int)
+  : LuceneRDDResponsePartition
+
   /**
    * Restricts the entries to those satisfying a predicate
    * @param pred Predicate to filter on

src/main/scala/org/zouzias/spark/lucenerdd/partition/LuceneRDDPartition.scala

@@ -145,6 +145,14 @@ private[lucenerdd] class LuceneRDDPartition[T]
       facetField + FacetedLuceneRDD.FacetTextFieldSuffix,
       topK)(Analyzer)
   }
+
+  override def moreLikeThis(fieldName: String, query: String,
+                            minTermFreq: Int, minDocFreq: Int, topK: Int)
+  : LuceneRDDResponsePartition = {
+    val docs = LuceneQueryHelpers.moreLikeThis(indexSearcher, fieldName,
+      query, minTermFreq, minDocFreq, topK)(Analyzer)
+    LuceneRDDResponsePartition(docs)
+  }
 }
 
 object LuceneRDDPartition {

src/main/scala/org/zouzias/spark/lucenerdd/query/LuceneQueryHelpers.scala

@@ -25,6 +25,7 @@ import org.apache.lucene.facet.{FacetsCollector, FacetsConfig}
 import org.apache.lucene.facet.sortedset.SortedSetDocValuesFacetCounts
 import org.apache.lucene.facet.taxonomy.{FastTaxonomyFacetCounts, TaxonomyReader}
 import org.apache.lucene.index.Term
+import org.apache.lucene.queries.mlt.MoreLikeThis
 import org.apache.lucene.queryparser.classic.QueryParser
 import org.apache.lucene.search._
 import org.zouzias.spark.lucenerdd.aggregate.SparkFacetResultMonoid
@@ -280,4 +281,29 @@ object LuceneQueryHelpers extends Serializable {
 
     searchTopK(indexSearcher, builder.build(), topK)
   }
+
+  /**
+    * Lucene's More Like This (MLT) functionality
+    * @param indexSearcher
+    * @param fieldName
+    * @param query
+    * @param minTermFreq
+    * @param minDocFreq
+    * @param topK
+    * @param analyzer
+    * @return
+    */
+  def moreLikeThis(indexSearcher: IndexSearcher, fieldName: String,
+                   query: String,
+                   minTermFreq: Int, minDocFreq: Int, topK: Int)
+                  (implicit analyzer: Analyzer)
+  : Iterator[SparkScoreDoc] = {
+    val mlt = new MoreLikeThis(indexSearcher.getIndexReader)
+    mlt.setMinTermFreq(minTermFreq)
+    mlt.setMinDocFreq(minDocFreq)
+    mlt.setFieldNames(Array(fieldName)) // FIXME: Is this necessary?
+    mlt.setAnalyzer(analyzer)
+    val q = mlt.like(fieldName, new StringReader(query))
+    searchTopK(indexSearcher, q, topK).toIterator
+  }
 }