diff --git a/.travis.yml b/.travis.yml
index 5c609610..c1ae7931 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,4 +1,6 @@
 language: scala
+scala:
+  - 2.11.12
 sudo: false
 dist: trusty
 cache:
@@ -15,22 +17,16 @@ before_cache:
 matrix:
   include:
     - jdk: oraclejdk8
-      scala: 2.11.12
-      env: TEST_SPARK_VERSION="2.4.0" LUCENE_ANALYZER="en" LINKER_METHOD="cartesian"
+      env: TEST_SPARK_VERSION="2.4.1" LUCENERDD_ANALYZER_NAME="en" LUCENERDD_LINKER_METHOD="cartesian"
     - jdk: openjdk8
-      scala: 2.11.12
-      env: TEST_SPARK_VERSION="2.4.0" LUCENE_ANALYZER="en" LINKER_METHOD="collectbroadcast"
+      env: TEST_SPARK_VERSION="2.4.1" LUCENERDD_ANALYZER_NAME="en" LUCENERDD_LINKER_METHOD="collectbroadcast"
     - jdk: openjdk8
-      scala: 2.11.12
-      env: TEST_SPARK_VERSION="2.4.0" LUCENE_ANALYZER="whitespace" LINKER_METHOD="cartesian"
+      env: TEST_SPARK_VERSION="2.4.1" LUCENERDD_ANALYZER_NAME="whitespace" LUCENERDD_LINKER_METHOD="cartesian"
     - jdk: oraclejdk8
-      scala: 2.11.12
-      env: TEST_SPARK_VERSION="2.4.0" LUCENE_ANALYZER="whitespace" LINKER_METHOD="collectbroadcast"
+      env: TEST_SPARK_VERSION="2.4.1" LUCENERDD_ANALYZER_NAME="whitespace" LUCENERDD_LINKER_METHOD="collectbroadcast"
 script:
   - sbt ++$TRAVIS_SCALA_VERSION clean update
-      -Dlucenerdd.linker.method=${LINKER_METHOD}
-      -Dlucenerdd.spatial.linker.method=${LINKER_METHOD}
-      -Dlucenerdd.analyzer.name=${LUCENE_ANALYZER} -test
+      -Dlucenerdd.spatial.linker.method=${LUCENE_SPATIAL_LINKER_METHOD} -test
   - sbt ++$TRAVIS_SCALA_VERSION scalastyle
   - sbt ++$TRAVIS_SCALA_VERSION assembly
   - travis_wait 30 sbt ++$TRAVIS_SCALA_VERSION clean coverage test coverageReport
diff --git a/README.md b/README.md
index 0fb6a2df..d3684de3 100644
--- a/README.md
+++ b/README.md
@@ -9,7 +9,7 @@
 
 [Spark](http://spark.apache.org) RDD with Apache [Lucene](https://lucene.apache.org)'s query capabilities.
 
-The main abstractions are special types of `RDD` called `LuceneRDD`, `FacetedLuceneRDD` and `ShapeLuceneRDD`, which instantiate a Lucene index on each Spark executor. These `RDD`s distribute search queries and aggregate search results between the Spark driver and its executors. Currently, the following queries are supported:
+The main abstractions are special types of `RDD` called `LuceneRDD` and `ShapeLuceneRDD`, which instantiate a Lucene index on each Spark executor. These `RDD`s distribute search queries and aggregate search results between the Spark driver and its executors. Currently, the following queries are supported:
 
 |Operation| Syntax| Description |
 |-------|---------------------|----------|
@@ -18,7 +18,6 @@ The main abstractions are special types of `RDD` called `LuceneRDD`, `FacetedLuc
 |Phrase Query | `LuceneRDD.phraseQuery(field, query, topK)` | Phrase search |
 |Prefix Query | `LuceneRDD.prefixSearch(field, prefix, topK)` | Prefix search |
 |Query Parser | `LuceneRDD.query(queryString, topK)` | Query parser search|
-|Faceted Search| `FacetedLuceneRDD.facetQuery(queryString, field, topK)` | Faceted Search |
 |[Record Linkage](https://github.com/zouzias/spark-lucenerdd/wiki/Record-Linkage)| `LuceneRDD.link(otherEntity: RDD[T], linkageFct: T => searchQuery, topK)`| Record linkage via Lucene queries|
 |Circle Search| `ShapeLuceneRDD.circleSearch((x,y), radius, topK)` | Search within radius |
 |Bbox Search| `ShapeLuceneRDD.bboxSearch(lowerLeft, upperLeft, topK)` | Bounding box |
diff --git a/build.sbt b/build.sbt
index b638fe83..94b4348b 100644
--- a/build.sbt
+++ b/build.sbt
@@ -77,10 +77,12 @@ pomExtra := <scm>
     </developer>
   </developers>
 
-val luceneV = "7.7.1"
+credentials += Credentials(Path.userHome / ".sbt" / ".credentials")
+
+val luceneV = "8.0.0"
 
 spName := "zouzias/spark-lucenerdd"
-sparkVersion := "2.4.0"
+sparkVersion := "2.4.1"
 spShortDescription := "Spark RDD with Lucene's query capabilities"
 sparkComponents ++= Seq("core", "sql", "mllib")
 spAppendScalaVersion := true
@@ -94,8 +96,8 @@ testSparkVersion := sys.props.get("spark.testVersion").getOrElse(sparkVersion.va
 
 
 // scalastyle:off
-val scalactic                 = "org.scalactic"                  %% "scalactic"                % "3.0.6"
-val scalatest                 = "org.scalatest"                  %% "scalatest"                % "3.0.6" % "test"
+val scalactic                 = "org.scalactic"                  %% "scalactic"                % "3.0.7"
+val scalatest                 = "org.scalatest"                  %% "scalatest"                % "3.0.7" % "test"
 
 val joda_time                 = "joda-time"                      % "joda-time"                 % "2.10.1"
 val algebird                  = "com.twitter"                    %% "algebird-core"            % "0.13.5"
@@ -104,7 +106,6 @@ val spatial4j                 = "org.locationtech.spatial4j"     % "spatial4j"
 
 val typesafe_config           = "com.typesafe"                   % "config"                    % "1.3.3"
 
-val lucene_facet              = "org.apache.lucene"              % "lucene-facet"              % luceneV
 val lucene_analyzers          = "org.apache.lucene"              % "lucene-analyzers-common"   % luceneV
 val lucene_query_parsers      = "org.apache.lucene"              % "lucene-queryparser"        % luceneV
 val lucene_expressions        = "org.apache.lucene"              % "lucene-expressions"        % luceneV
@@ -117,7 +118,6 @@ val jts                       = "org.locationtech.jts"           % "jts-core"
 
 libraryDependencies ++= Seq(
   algebird,
-  lucene_facet,
   lucene_analyzers,
   lucene_expressions,
   lucene_query_parsers,
diff --git a/project/build.properties b/project/build.properties
index 133a8f19..8e682c52 100644
--- a/project/build.properties
+++ b/project/build.properties
@@ -1 +1 @@
-sbt.version=0.13.17
+sbt.version=0.13.18
diff --git a/project/buildinfo.sbt b/project/buildinfo.sbt
deleted file mode 100644
index b97dffd9..00000000
--- a/project/buildinfo.sbt
+++ /dev/null
@@ -1,18 +0,0 @@
-/*
- * 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.
- */
-addSbtPlugin("com.eed3si9n" % "sbt-buildinfo" % "0.7.0")
-
diff --git a/project/plugins.sbt b/project/plugins.sbt
index 88e25f34..e3447af7 100644
--- a/project/plugins.sbt
+++ b/project/plugins.sbt
@@ -17,11 +17,13 @@
 
 resolvers += "bintray-spark-packages" at "https://dl.bintray.com/spark-packages/maven/"
 
-addSbtPlugin("com.timushev.sbt" % "sbt-updates" % "0.3.0")
+addSbtPlugin("com.eed3si9n" % "sbt-buildinfo" % "0.7.0")
 
-addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.7")
+addSbtPlugin("com.timushev.sbt" % "sbt-updates" % "0.4.0")
 
-addSbtPlugin("com.github.gseitz" % "sbt-release" % "1.0.9")
+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.9")
+
+addSbtPlugin("com.github.gseitz" % "sbt-release" % "1.0.11")
 
 addSbtPlugin("org.scalastyle" %% "scalastyle-sbt-plugin" % "1.0.0")
 
diff --git a/src/main/resources/reference.conf b/src/main/resources/reference.conf
index 5c28900b..e0003f36 100644
--- a/src/main/resources/reference.conf
+++ b/src/main/resources/reference.conf
@@ -2,15 +2,12 @@ lucenerdd {
 
   // Name of analyzer as it is under Lucene's package org.apache.lucene.analysis.XX
   analyzer.name = "en"
+  analyzer.name=${?LUCENERDD_ANALYZER_NAME}
 
-  // Analyzer name must be "ngram"
-  analyzer {
-    ngram.mingram = 2
-    ngram.maxgram = 5
-  }
-
+  
   // Similarity scoring for Lucenes
   similarity.name = "bm25" // anything else will default to Lucene classic similarity
+  similarity.name = ${?LUCENERDD_SIMILARITY_NAME}
 
   // Supported linkage methods
   // "collectbroadcast" : Collects the RDD that contains the queries (to be used only if query RDD
@@ -20,6 +17,7 @@ lucenerdd {
   // of LuceneRDD. Note it duplicates each partition of LuceneRDD n times where n is the number of
   // partitions of the queries RDD.
   linker.method = "collectbroadcast"
+  linker.method = ${?LUCENERDD_LINKER_METHOD}
 
   index {
 
@@ -28,17 +26,20 @@ lucenerdd {
     // Otherwise the index will be stored in memory
     // Do not use memory, see http://lucene.apache.org/core/7_5_0/core/org/apache/lucene/store/RAMDirectory.html
     store.mode = "disk"
+    store.mode = ${?LUCENERDD_INDEX_STORE_MODE}
 
     stringfields{
 
       // Analyze string fields by default or not
       // Implicit fields, like _1, _2, etc will use this option
       analyzed = true
+      analyzed = ${?LUCENERDD_INDEX_STRINGFIELDS_ANALYZED}
 
       // Select a subset of string fields that you do not wish to be analyzed
       // Due to serialization issues this list should be set before starting a Spark Session
       // Moreover, all text/string fields that end with '_notanalyzed' are not analyzed
       not_analyzed_list = []
+      not_analyzed_list = ${?LUCENERDD_INDEX_STRINGFIELDS_NOT_ANALYZED_LIST}
 
       // Text fields options as in org.apache.lucene.index.IndexOptions
       //
@@ -49,36 +50,55 @@ lucenerdd {
       // "DOCS_AND_FREQS_AND_POSITIONS_AND_OFFSETS"
       // "NONE"
       options = "docs_and_freqs_and_positions_and_offsets"
+      options = ${?LUCENERDD_INDEX_STRINGFIELDS_OPTIONS}
 
-      // Omit terms norms
-      terms.omitnorms = false
+      terms {
+        // Omit terms norms
+        omitnorms = false
+        omitnorms = ${?LUCENERDD_INDEX_STRINGFIELDS_TERMS_OMITNORMS}
 
-      // Store term positions
-      terms.positions = false
+        // Store term positions
+        positions = false
+        positions = ${?LUCENERDD_INDEX_STRINGFIELDS_TERMS_POSITIONS}
 
-      // Store Term vectors (set true, otherwise LuceneRDD.termVectors(fieldName) will fail)
-      terms.vectors = true
+        // Store Term vectors (set true, otherwise LuceneRDD.termVectors(fieldName) will fail)
+        vectors = true
+        vectors = ${?LUCENERDD_INDEX_STRINGFIELDS_TERMS_VECTORS}
+      }
     }
   }
 
-  // Maximum value on topK queries
-  query.topk.maxvalue = 100
-  // Default value of number of returned results
-  query.topk.default = 10
+  query {
+    // Maximum value on topK queries
+    topk.maxvalue = 100
+    topk.maxvalue = ${?LUCENERDD_QUERY_TOPK_MAXVALUE}
 
-  // Default value of number of faceted results
-  query.facets.number.default = 10
+    // Default value of number of returned results
+    topk.default = 10
+    topk.default = ${?LUCENERDD_QUERY_TOPK_DEFAULT}
+
+  }
 
   // Spatial related configurations used by ShapeLuceneRDD
   spatial {
     prefixtree {
+
+      // Spatial tree data structure
       name = "quad"  // "geohash" or "quad"
+      name = ${?LUCENE_SPATIAL_PREFIXTREE_NAME}
+
       maxlevel = 9 // 11 results in sub-meter precision for geohash
+      maxlevel = ${?LUCENE_SPATIAL_PREFIXTREE_MAXLEVEL}
+
+
       maxDistErr = 5.0 // in kilometers
+      maxDistErr = ${?LUCENE_SPATIAL_PREFIXTREE_MAXDISTERR}
+
     }
 
     // Shape format can be one of ShapeIO.GeoJSON, ShapeIO.LEGACY, ShapeIO.POLY, ShapeIO.WKT
     shape.io.format = "WKT"
+    shape.io.format = ${?LUCENE_SPATIAL_SHAPE_IO_FORMAT}
 
     // Supported linkage methods
     // "collectbroadcast" : Collects the RDD that contains the queries (to be used only if query RDD
@@ -88,5 +108,6 @@ lucenerdd {
     // of LuceneRDD. Note it duplicates each partition of LuceneRDD n times where n is the number of
     // partitions of the queries RDD.
     linker.method = "collectbroadcast"
+    linker.method = ${?LUCENE_SPATIAL_LINKER_METHOD}
   }
 }
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/LuceneRDD.scala b/src/main/scala/org/zouzias/spark/lucenerdd/LuceneRDD.scala
index 780e7074..976a704c 100644
--- a/src/main/scala/org/zouzias/spark/lucenerdd/LuceneRDD.scala
+++ b/src/main/scala/org/zouzias/spark/lucenerdd/LuceneRDD.scala
@@ -19,7 +19,7 @@ package org.zouzias.spark.lucenerdd
 
 import com.twitter.algebird.{TopK, TopKMonoid}
 import org.apache.lucene.document.Document
-import org.zouzias.spark.lucenerdd.config.LuceneRDDConfigurable
+import org.zouzias.spark.lucenerdd.config.{LuceneRDDConfigurable, LuceneRDDParams}
 import org.zouzias.spark.lucenerdd.response.{LuceneRDDResponse, LuceneRDDResponsePartition}
 import org.apache.spark.rdd.RDD
 import org.apache.lucene.search.Query
@@ -48,6 +48,8 @@ import scala.reflect.ClassTag
 class LuceneRDD[T: ClassTag](protected val partitionsRDD: RDD[AbstractLuceneRDDPartition[T]],
                              protected val indexAnalyzer: String,
                              protected val queryAnalyzer: String,
+                             protected val indexAnalyzerPerField: Map[String, String],
+                             protected val queryAnalyzerPerField: Map[String, String],
                              protected val similarity: String)
   extends RDD[T](partitionsRDD.context, List(new OneToOneDependency(partitionsRDD)))
   with LuceneRDDConfigurable {
@@ -356,7 +358,8 @@ class LuceneRDD[T: ClassTag](protected val partitionsRDD: RDD[AbstractLuceneRDDP
     val newPartitionRDD = partitionsRDD.mapPartitions(partition =>
       partition.map(_.filter(pred)), preservesPartitioning = true
     )
-    new LuceneRDD(newPartitionRDD, indexAnalyzer, queryAnalyzer, similarity)
+    new LuceneRDD(newPartitionRDD, indexAnalyzer, queryAnalyzer,
+      indexAnalyzerPerField, queryAnalyzerPerField, similarity)
   }
 
   def exists(elem: T): Boolean = {
@@ -377,37 +380,38 @@ object LuceneRDD extends Versionable
    * Instantiate a LuceneRDD given an RDD[T]
    *
    * @param elems RDD of type T
-   * @param indexAnalyzer Index Analyzer name
-   * @param queryAnalyzer Query Analyzer name
+   * @param indexAnalyzer Index analyzer name
+   * @param queryAnalyzer Query analyzer name
    * @param similarity Lucene scoring similarity, i.e., BM25 or TF-IDF
+   * @param indexAnalyzerPerField Lucene Analyzer per field (indexing time), default empty
+   * @param queryAnalyzerPerField Lucene Analyzer per field (query time), default empty
    * @tparam T Generic type
    * @return
    */
   def apply[T : ClassTag](elems: RDD[T],
                           indexAnalyzer: String,
                           queryAnalyzer: String,
-                          similarity: String)
+                          similarity: String,
+                          indexAnalyzerPerField: Map[String, String],
+                          queryAnalyzerPerField: Map[String, String])
     (implicit conv: T => Document): LuceneRDD[T] = {
     val partitions = elems.mapPartitionsWithIndex[AbstractLuceneRDDPartition[T]](
       (partId, iter) => Iterator(LuceneRDDPartition(iter, partId, indexAnalyzer, queryAnalyzer,
         similarity)),
       preservesPartitioning = true)
-    new LuceneRDD[T](partitions, indexAnalyzer, queryAnalyzer, similarity)
-  }
-
-  def apply[T : ClassTag](elems: RDD[T])
-                         (implicit conv: T => Document): LuceneRDD[T] = {
-   apply(elems, getOrElseEn(IndexAnalyzerConfigName), getOrElseEn(QueryAnalyzerConfigName),
-     getOrElseClassic())
+    new LuceneRDD[T](partitions, indexAnalyzer, queryAnalyzer,
+      indexAnalyzerPerField, queryAnalyzerPerField, similarity)
   }
 
   /**
    * Instantiate a LuceneRDD with an iterable
    *
    * @param elems Elements to index
-   * @param indexAnalyzer Index Analyzer name
-   * @param queryAnalyzer Query Analyzer name
+   * @param indexAnalyzer Index analyzer name
+   * @param queryAnalyzer Query analyzer name
    * @param similarity Lucene scoring similarity, i.e., BM25 or TF-IDF
+   * @param indexAnalyzerPerField Lucene Analyzer per field (indexing time), default empty
+   * @param queryAnalyzerPerField Lucene Analyzer per field (query time), default empty
    * @param sc Spark Context
    * @tparam T Input type
    * @return
@@ -416,12 +420,28 @@ object LuceneRDD extends Versionable
   (elems: Iterable[T],
    indexAnalyzer: String,
    queryAnalyzer: String,
-   similarity: String)
+   similarity: String,
+   indexAnalyzerPerField: Map[String, String],
+   queryAnalyzerPerField: Map[String, String])
   (implicit sc: SparkContext, conv: T => Document)
   : LuceneRDD[T] = {
-    apply[T](sc.parallelize[T](elems.toSeq), indexAnalyzer, queryAnalyzer, similarity)
+    apply[T](sc.parallelize[T](elems.toSeq), indexAnalyzer, queryAnalyzer, similarity,
+      indexAnalyzerPerField, queryAnalyzerPerField)
+  }
+
+  def apply[T : ClassTag]
+  (elems: RDD[T])
+  (implicit conv: T => Document)
+  : LuceneRDD[T] = {
+    apply[T](elems,
+      getOrElseEn(IndexAnalyzerConfigName),
+      getOrElseEn(QueryAnalyzerConfigName),
+      getOrElseClassic(),
+      Map.empty[String, String],
+      Map.empty[String, String])
   }
 
+
   def apply[T : ClassTag]
   (elems: Iterable[T])
   (implicit sc: SparkContext, conv: T => Document)
@@ -429,24 +449,35 @@ object LuceneRDD extends Versionable
     apply[T](elems,
       getOrElseEn(IndexAnalyzerConfigName),
       getOrElseEn(QueryAnalyzerConfigName),
-      getOrElseClassic())
+      getOrElseClassic(),
+      Map.empty[String, String],
+      Map.empty[String, String])
   }
 
   /**
-   * Instantiate a LuceneRDD with DataFrame
+   * Instantiate a LuceneRDD from a  [[DataFrame]]
    *
-   * @param dataFrame Spark DataFrame
-   * @param indexAnalyzer Index Analyzer name
-   * @param queryAnalyzer Query Analyzer name
-   * @param similarity Lucene scoring similarity, i.e., BM25 or TF-IDF
-    * @return
+   * @param dataFrame Spark [[DataFrame]]
+   * @return
    */
+  def apply(dataFrame: DataFrame,
+            indexAnalyzer: String,
+            queryAnalyzer: String,
+            similarity: String,
+            indexAnalyzerPerField: Map[String, String],
+            queryAnalyzerPerField: Map[String, String])
+  : LuceneRDD[Row] = {
+    apply[Row](dataFrame.rdd, indexAnalyzer, queryAnalyzer, similarity,
+      indexAnalyzerPerField, queryAnalyzerPerField)
+  }
+
   def apply(dataFrame: DataFrame,
             indexAnalyzer: String,
             queryAnalyzer: String,
             similarity: String)
   : LuceneRDD[Row] = {
-    apply[Row](dataFrame.rdd, indexAnalyzer, queryAnalyzer, similarity)
+    apply[Row](dataFrame.rdd, indexAnalyzer, queryAnalyzer, similarity,
+      Map.empty[String, String], Map.empty[String, String])
   }
 
   /**
@@ -460,7 +491,9 @@ object LuceneRDD extends Versionable
     apply[Row](dataFrame.rdd,
       getOrElseEn(IndexAnalyzerConfigName),
       getOrElseEn(QueryAnalyzerConfigName),
-      getOrElseClassic())
+      getOrElseClassic(),
+      Map.empty[String, String],
+      Map.empty[String, String])
   }
 
   /**
@@ -473,9 +506,7 @@ object LuceneRDD extends Versionable
     * @param queryPartColumns List of query columns for [[HashPartitioner]]
     * @param entityPartColumns List of entity columns for [[HashPartitioner]]
     * @param topK Number of linked results
-    * @param indexAnalyzer Lucene analyzer at index time
-    * @param queryAnalyzer Lucene analyzer at query time
-    * @param similarity Lucene Similarity metric (BM25, Tf/idf)
+    * @param luceneRDDParams Parameters for index and query time analysis
     * @return Returns top-k linked results as RDD of [[Tuple2]] where _1 is query and
     *         _2 is top-k linked results as [[SparkScoreDoc]].
     */
@@ -485,9 +516,7 @@ object LuceneRDD extends Versionable
                          queryPartColumns: Array[String],
                          entityPartColumns: Array[String],
                          topK : Int = 3,
-                         indexAnalyzer: String = getOrElseEn(IndexAnalyzerConfigName),
-                         queryAnalyzer: String = getOrElseEn(QueryAnalyzerConfigName),
-                         similarity: String = getOrElseClassic())
+                         luceneRDDParams: LuceneRDDParams = LuceneRDDParams())
   : RDD[(Row, Array[SparkScoreDoc])] = {
 
     assert(entityPartColumns.nonEmpty,
@@ -517,8 +546,12 @@ object LuceneRDD extends Versionable
         iterKeyedByHash.flatMap { case (_, (qs, ents)) =>
 
           // Instantiate a lucene index on partitioned entities
-          val lucenePart = LuceneRDDPartition(ents.toIterator, idx, indexAnalyzer,
-            queryAnalyzer, similarity)
+          val lucenePart = LuceneRDDPartition(ents.toIterator, idx,
+            luceneRDDParams.indexAnalyzer,
+            luceneRDDParams.queryAnalyzer,
+            luceneRDDParams.similarity,
+            luceneRDDParams.indexAnalyzerPerField,
+            luceneRDDParams.queryAnalyzerPerField)
 
           // Multi-query lucene index
           qs.map(q => (q, lucenePart.query(rowToQuery(q), topK).results.toArray))
@@ -533,9 +566,7 @@ object LuceneRDD extends Versionable
     * @param rowToQuery Function that maps [[Row]] to Lucene [[Query]]
     * @param blockingColumns Columns on which exact match is required
     * @param topK Number of top-K query results
-    * @param indexAnalyzer Lucene analyzer at index time
-    * @param queryAnalyzer Lucene analyzer at query time
-    * @param similarity Lucene Similarity metric (BM25, Tf/idf)
+    * @param luceneRDDParams Parameters for index-time and query-time analysis
     * @return Returns top-k deduplicated results as [[RDD]] of [[Tuple2]] where _1 is query and
     *         _2 is top-k linked results as [[SparkScoreDoc]].
     * @return
@@ -544,9 +575,7 @@ object LuceneRDD extends Versionable
                  rowToQuery: Row => Query,
                  blockingColumns: Array[String],
                  topK : Int = 3,
-                 indexAnalyzer: String = getOrElseEn(IndexAnalyzerConfigName),
-                 queryAnalyzer: String = getOrElseEn(QueryAnalyzerConfigName),
-                 similarity: String = getOrElseClassic())
+                 luceneRDDParams: LuceneRDDParams = LuceneRDDParams())
   : RDD[(Row, Array[SparkScoreDoc])] = {
 
     // Check that there is at least one partition column
@@ -569,10 +598,12 @@ object LuceneRDD extends Versionable
       val (iterQueries, iterLucene) = iterKeyedByHash.map(_._2).duplicate
 
       // Instantiate a lucene index on partitioned entities
-      val lucenePart = LuceneRDDPartition(iterLucene,
-        idx,
-        indexAnalyzer,
-        queryAnalyzer, similarity)
+      val lucenePart = LuceneRDDPartition(iterLucene, idx,
+        luceneRDDParams.indexAnalyzer,
+        luceneRDDParams.queryAnalyzer,
+        luceneRDDParams.similarity,
+        luceneRDDParams.indexAnalyzerPerField,
+        luceneRDDParams.queryAnalyzerPerField)
 
       // Multi-query lucene index
       iterQueries.map(q => (q, lucenePart.query(rowToQuery(q), topK).results.toArray))
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/LuceneRDDKryoRegistrator.scala b/src/main/scala/org/zouzias/spark/lucenerdd/LuceneRDDKryoRegistrator.scala
index 0f7a6e54..06cb51cd 100644
--- a/src/main/scala/org/zouzias/spark/lucenerdd/LuceneRDDKryoRegistrator.scala
+++ b/src/main/scala/org/zouzias/spark/lucenerdd/LuceneRDDKryoRegistrator.scala
@@ -20,8 +20,7 @@ import com.twitter.algebird.TopK
 import com.twitter.chill.Kryo
 import org.apache.spark.SparkConf
 import org.apache.spark.serializer.{KryoRegistrator, KryoSerializer}
-import org.zouzias.spark.lucenerdd.facets.FacetedLuceneRDD
-import org.zouzias.spark.lucenerdd.models.{SparkDoc, SparkFacetResult, SparkScoreDoc}
+import org.zouzias.spark.lucenerdd.models.{SparkDoc, SparkScoreDoc}
 import org.zouzias.spark.lucenerdd.partition.LuceneRDDPartition
 import org.zouzias.spark.lucenerdd.response.{LuceneRDDResponse, LuceneRDDResponsePartition}
 import org.zouzias.spark.lucenerdd.testing.{FavoriteCaseClass, Person}
@@ -30,7 +29,6 @@ class LuceneRDDKryoRegistrator extends KryoRegistrator {
   def registerClasses(kryo: Kryo): Unit = {
     kryo.register(classOf[LuceneRDD[_]])
     kryo.register(classOf[LuceneRDDPartition[_]])
-    kryo.register(classOf[FacetedLuceneRDD[_]])
     kryo.register(classOf[SparkDoc])
     kryo.register(classOf[Number])
     kryo.register(classOf[java.lang.Double])
@@ -57,7 +55,6 @@ class LuceneRDDKryoRegistrator extends KryoRegistrator {
     kryo.register(classOf[scala.collection.immutable.Set$EmptySet$])
     kryo.register(classOf[scala.collection.immutable.Map[_, _]])
     kryo.register(classOf[Array[scala.collection.immutable.Map[_, _]]])
-    kryo.register(classOf[SparkFacetResult])
     kryo.register(classOf[SparkScoreDoc])
     kryo.register(classOf[LuceneRDDResponse])
     kryo.register(classOf[LuceneRDDResponsePartition])
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/aggregate/SparkFacetResultMonoid.scala b/src/main/scala/org/zouzias/spark/lucenerdd/aggregate/SparkFacetResultMonoid.scala
deleted file mode 100644
index 3de0cb7e..00000000
--- a/src/main/scala/org/zouzias/spark/lucenerdd/aggregate/SparkFacetResultMonoid.scala
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * 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.
- */
-package org.zouzias.spark.lucenerdd.aggregate
-
-import com.twitter.algebird.MapMonoid
-import org.zouzias.spark.lucenerdd.models.SparkFacetResult
-
-/**
- * Monoid used to aggregate faceted results [[SparkFacetResult]]
- * from the executors to the driver
- */
-object SparkFacetResultMonoid extends Serializable {
-
-  private lazy val facetMonoid = new MapMonoid[String, Long]()
-
-  def zero(facetName: String): SparkFacetResult = SparkFacetResult(facetName, facetMonoid.zero)
-
-  def plus(l: SparkFacetResult, r: SparkFacetResult): SparkFacetResult = {
-    require(l.facetName == r.facetName) // Check if summing same facets
-    SparkFacetResult(l.facetName, facetMonoid.plus(l.facets, r.facets))
-  }
-}
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/analyzers/AnalyzerConfigurable.scala b/src/main/scala/org/zouzias/spark/lucenerdd/analyzers/AnalyzerConfigurable.scala
index e1135b84..e80ac7b3 100644
--- a/src/main/scala/org/zouzias/spark/lucenerdd/analyzers/AnalyzerConfigurable.scala
+++ b/src/main/scala/org/zouzias/spark/lucenerdd/analyzers/AnalyzerConfigurable.scala
@@ -49,29 +49,26 @@ import org.apache.lucene.analysis.ru.RussianAnalyzer
 import org.apache.lucene.analysis.standard.StandardAnalyzer
 import org.apache.lucene.analysis.tr.TurkishAnalyzer
 import org.zouzias.spark.lucenerdd.config.Configurable
+import org.zouzias.spark.lucenerdd.logging.Logging
 
 /**
- * Lucene Analyzer loader via configuration
+ * Lucene Analyzer loader via configuration or via class name
+  *
+  * An analyzer can be loaded by using all the short country codes, i.e.,
+  * en,el,de, etc or using a class name present in the classpath, i.e.,
+  * 'org.apache.lucene.analysis.el.GreekAnalyzer'
+  *
+  * Custom Analyzers can be loaded provided that are present during runtime.
  */
-trait AnalyzerConfigurable extends Configurable {
+trait AnalyzerConfigurable extends Configurable
+  with Logging {
 
   private val IndexAnalyzerConfigKey = "lucenerdd.index.analyzer.name"
   private val QueryAnalyzerConfigKey = "lucenerdd.query.analyzer.name"
-  private val NgramMinGramConfigKey = "lucenerdd.index.analyzer.ngram.mingram"
-  private val NgramMaxGramConfigKey = "lucenerdd.index.analyzer.ngram.maxgram"
-
 
   /** Get the configured analyzers or fallback to English */
   protected def getOrElseEn(analyzerName: Option[String]): String = analyzerName.getOrElse("en")
 
-  private val NgramMinGram = if (Config.hasPath(NgramMinGramConfigKey)) {
-    Config.getInt(NgramMinGramConfigKey)
-  } else 2
-
-  private val NgramMaxGram = if (Config.hasPath(NgramMaxGramConfigKey)) {
-    Config.getInt(NgramMaxGramConfigKey)
-  } else NgramMinGram + 2 // Default is + 2
-
   protected val IndexAnalyzerConfigName: Option[String] =
     if (Config.hasPath(IndexAnalyzerConfigKey)) {
     Some(Config.getString(IndexAnalyzerConfigKey))} else None
@@ -113,12 +110,44 @@ trait AnalyzerConfigurable extends Configurable {
         case "pt" => new PortugueseAnalyzer()
         case "ru" => new RussianAnalyzer()
         case "tr" => new TurkishAnalyzer()
-        case "ngram" => new NgramAnalyzer(NgramMinGram, NgramMaxGram) // Example of custom analyzer
-        case _ => new StandardAnalyzer()
+        case otherwise: String =>
+          try {
+            val clazz = loadConstructor[Analyzer](otherwise)
+            clazz
+          }
+          catch {
+            case e: ClassNotFoundException =>
+              logError(s"Class ${otherwise} was not found in classpath. Does the class exist?", e)
+              null
+            case e: ClassCastException =>
+              logError(s"Class ${otherwise} could not be " +
+                s"cast to superclass org.apache.lucene.analysis.Analyzer.", e)
+              null
+            case e: Throwable =>
+              logError(s"Class ${otherwise} could not be used as Analyzer.", e)
+              null
+          }
       }
     }
     else {
+      logInfo("Analyzer name is not defined. Default analyzer is StandardAnalyzer().")
       new StandardAnalyzer()
     }
   }
+
+  /**
+    * Load a Lucene [[Analyzer]] using class name
+    *
+    * @param className The class name of the analyzer to load
+    * @tparam T
+    * @return Returns a Lucene Analyzer
+    */
+  private def loadConstructor[T <: Analyzer](className: String): T = {
+    val loader = getClass.getClassLoader
+    logInfo(s"Loading class ${className} using loader ${loader}")
+    val loadedClass: Class[T] = loader.loadClass(className).asInstanceOf[Class[T]]
+    val constructor = loadedClass.getConstructor()
+    constructor.newInstance()
+  }
+
 }
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/analyzers/NgramAnalyzer.scala b/src/main/scala/org/zouzias/spark/lucenerdd/analyzers/NgramAnalyzer.scala
deleted file mode 100644
index 0bbe6a14..00000000
--- a/src/main/scala/org/zouzias/spark/lucenerdd/analyzers/NgramAnalyzer.scala
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.
- */
-package org.zouzias.spark.lucenerdd.analyzers
-
-import org.apache.lucene.analysis.{Analyzer, Tokenizer}
-import org.apache.lucene.analysis.Analyzer.TokenStreamComponents
-import org.apache.lucene.analysis.core.{LowerCaseFilter, WhitespaceTokenizer}
-import org.apache.lucene.analysis.ngram.NGramTokenFilter
-
-/**
-  * An example with a custom Lucene analyzer (whitespace tokenizer, lowerCase
-  * and Ngram tokenizer filter)
-  * Creates NGramTokenizer with given min and max n-grams.
-  *
-  * @param minGram the smallest n-gram to generate
-  * @param maxGram the largest n-gram to generate
-  */
-class NgramAnalyzer(minGram: Int, maxGram: Int) extends Analyzer {
-  override def createComponents(fieldName: String): TokenStreamComponents = {
-    val source: Tokenizer = new WhitespaceTokenizer()
-    val lowerCase = new LowerCaseFilter(source)
-    val ngramSource = new NGramTokenFilter(lowerCase, minGram, maxGram)
-    new TokenStreamComponents(source, ngramSource)
-  }
-}
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/config/LuceneRDDConfigurable.scala b/src/main/scala/org/zouzias/spark/lucenerdd/config/LuceneRDDConfigurable.scala
index 23ea7185..2a850ce6 100644
--- a/src/main/scala/org/zouzias/spark/lucenerdd/config/LuceneRDDConfigurable.scala
+++ b/src/main/scala/org/zouzias/spark/lucenerdd/config/LuceneRDDConfigurable.scala
@@ -39,13 +39,6 @@ trait LuceneRDDConfigurable extends Configurable {
     else 10
   }
 
-  protected val DefaultFacetNum: Int = {
-    if (Config.hasPath("lucenerdd.query.facet.topk.default")) {
-      Config.getInt("lucenerdd.query.facet.topk.default")
-    }
-    else 10
-  }
-
   protected val StringFieldsDefaultAnalyzed: Boolean = {
     if (Config.hasPath("lucenerdd.index.stringfields.analyzed")) {
       Config.getBoolean("lucenerdd.index.stringfields.analyzed")
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/config/LuceneRDDParams.scala b/src/main/scala/org/zouzias/spark/lucenerdd/config/LuceneRDDParams.scala
new file mode 100644
index 00000000..a46e0260
--- /dev/null
+++ b/src/main/scala/org/zouzias/spark/lucenerdd/config/LuceneRDDParams.scala
@@ -0,0 +1,46 @@
+/*
+ * 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.
+ */
+package org.zouzias.spark.lucenerdd.config
+
+import org.apache.lucene.analysis.Analyzer
+import org.zouzias.spark.lucenerdd.analyzers.AnalyzerConfigurable
+import org.zouzias.spark.lucenerdd.query.SimilarityConfigurable
+
+ /** Lucene analysis parameters during indexing and querying.
+   *
+   * @param indexAnalyzer         Index analyzer name. Lucene [[Analyzer]] used during indexing
+   * @param queryAnalyzer         Query analyzer name. Lucene [[Analyzer]] used during querying
+   * @param similarity            Lucene scoring similarity, i.e., BM25 or TF-IDF
+   * @param indexAnalyzerPerField Lucene Analyzer per field (indexing time), default empty
+   * @param queryAnalyzerPerField Lucene Analyzer per field (query time), default empty
+   */
+case class LuceneRDDParams(indexAnalyzer: String,
+                           queryAnalyzer: String,
+                           similarity: String,
+                           indexAnalyzerPerField: Map[String, String],
+                           queryAnalyzerPerField: Map[String, String]) extends Serializable
+
+
+object LuceneRDDParams extends AnalyzerConfigurable with SimilarityConfigurable {
+  def apply(): LuceneRDDParams = {
+    new LuceneRDDParams(getOrElseEn(IndexAnalyzerConfigName),
+      getOrElseEn(QueryAnalyzerConfigName),
+      getOrElseClassic(),
+      Map.empty[String, String],
+      Map.empty[String, String])
+  }
+}
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/facets/FacetedLuceneRDD.scala b/src/main/scala/org/zouzias/spark/lucenerdd/facets/FacetedLuceneRDD.scala
deleted file mode 100644
index c1db740a..00000000
--- a/src/main/scala/org/zouzias/spark/lucenerdd/facets/FacetedLuceneRDD.scala
+++ /dev/null
@@ -1,191 +0,0 @@
-/*
- * 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.
- */
-package org.zouzias.spark.lucenerdd.facets
-
-import org.apache.lucene.document.Document
-import org.apache.spark.SparkContext
-import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.{DataFrame, Row}
-import org.apache.spark.storage.StorageLevel
-import org.zouzias.spark.lucenerdd.LuceneRDD
-import org.zouzias.spark.lucenerdd.aggregate.SparkFacetResultMonoid
-import org.zouzias.spark.lucenerdd.analyzers.AnalyzerConfigurable
-import org.zouzias.spark.lucenerdd.models.SparkFacetResult
-import org.zouzias.spark.lucenerdd.partition.{AbstractLuceneRDDPartition, LuceneRDDPartition}
-import org.zouzias.spark.lucenerdd.query.SimilarityConfigurable
-import org.zouzias.spark.lucenerdd.response.LuceneRDDResponse
-import org.zouzias.spark.lucenerdd.versioning.Versionable
-
-import scala.reflect.ClassTag
-
-/**
- * LuceneRDD with faceted functionality
- */
-class FacetedLuceneRDD[T: ClassTag]
-  (override protected val partitionsRDD: RDD[AbstractLuceneRDDPartition[T]],
-   override val indexAnalyzer: String,
-   override val queryAnalyzer: String,
-   override val similarity: String)
-  extends LuceneRDD[T](partitionsRDD, indexAnalyzer, queryAnalyzer, similarity) {
-
-  setName("FacetedLuceneRDD")
-
-  override def cache(): this.type = {
-    this.persist(StorageLevel.MEMORY_ONLY)
-  }
-
-  override def persist(newLevel: StorageLevel): this.type = {
-    partitionsRDD.persist(newLevel)
-    super.persist(newLevel)
-    this
-  }
-
-  override def unpersist(blocking: Boolean = true): this.type = {
-    partitionsRDD.unpersist(blocking)
-    super.unpersist(blocking)
-    this
-  }
-
-  /**
-   * Aggregates faceted search results using monoidal structure [[SparkFacetResultMonoid]]
-   *
-   * @param f a function that computes faceted search results per partition
-   * @return faceted search results
-   */
-  private def facetResultsAggregator(f: AbstractLuceneRDDPartition[T] => SparkFacetResult)
-  : SparkFacetResult = {
-    partitionsRDD.map(f(_)).reduce(SparkFacetResultMonoid.plus)
-  }
-
-  /**
-   * Faceted query
-   *
-   * @param searchString Lucene query string
-   * @param facetField Field on which to compute facet
-   * @param topK Number of results
-   * @param facetNum Number of faceted results
-   * @return
-   */
-  def facetQuery(searchString: String,
-                 facetField: String,
-                 topK: Int = DefaultTopK,
-                 facetNum: Int = DefaultFacetNum
-                ): (LuceneRDDResponse, SparkFacetResult) = {
-    val aggrTopDocs = partitionMapper(_.query(searchString, topK))
-    val aggrFacets = facetResultsAggregator(_.facetQuery(searchString, facetField, facetNum))
-    (aggrTopDocs, aggrFacets)
-  }
-
-  /**
-   * Faceted query with multiple facets
-   *
-   * @param searchString Lucene query string
-   * @param facetFields Fields on which to compute facets
-   * @param topK Number of results
-   * @param facetNum Number of faceted results
-   * @return
-   */
-  def facetQueries(searchString: String,
-                   facetFields: Seq[String],
-                   topK: Int = DefaultTopK,
-                   facetNum: Int = DefaultFacetNum)
-  : (LuceneRDDResponse, Map[String, SparkFacetResult]) = {
-    logInfo(s"Faceted query on facet fields ${facetFields.mkString(",")}...")
-    val aggrTopDocs = partitionMapper(_.query(searchString, topK))
-    val aggrFacets = facetFields.map { case facetField =>
-      (facetField, facetResultsAggregator(_.facetQuery(searchString, facetField, facetNum)))
-    }.toMap[String, SparkFacetResult]
-    (aggrTopDocs, aggrFacets)
-  }
-}
-
-object FacetedLuceneRDD extends Versionable
-  with AnalyzerConfigurable
-  with SimilarityConfigurable {
-
-  /** All faceted fields are suffixed with _facet */
-  val FacetTextFieldSuffix = "_facet"
-  val FacetNumericFieldSuffix = "_numFacet"
-
-  /**
-   * Instantiate a FacetedLuceneRDD given an RDD[T]
-   *
-   * @param elems RDD of type T
-   * @tparam T Generic type
-   * @return
-   */
-  def apply[T : ClassTag](elems: RDD[T], indexAnalyzer: String, queryAnalyzer: String,
-                          similarity: String)
-                         (implicit conv: T => Document): FacetedLuceneRDD[T] = {
-    val partitions = elems.mapPartitionsWithIndex[AbstractLuceneRDDPartition[T]](
-      (partId, iter) => Iterator(LuceneRDDPartition(iter, partId, indexAnalyzer, queryAnalyzer,
-        similarity)),
-      preservesPartitioning = true)
-    new FacetedLuceneRDD[T](partitions, indexAnalyzer, queryAnalyzer, similarity)
-  }
-
-  def apply[T : ClassTag](elems: RDD[T])(implicit conv: T => Document)
-  : FacetedLuceneRDD[T] = {
-    apply[T](elems, getOrElseEn(IndexAnalyzerConfigName), getOrElseEn(QueryAnalyzerConfigName),
-      getOrElseClassic())
-  }
-
-  /**
-   * Instantiate a FacetedLuceneRDD with an iterable
-   *
-   * @param elems Iterable of documents
-   * @param indexAnalyzer Index Analyzer name
-   * @param queryAnalyzer Query Analyzer name
-   * @param similarity Lucene scoring similarity, i.e., BM25 or TF-IDF
-   * @param sc
-   * @tparam T
-   * @return
-   */
-  def apply[T : ClassTag]
-  (elems: Iterable[T], indexAnalyzer: String, queryAnalyzer: String, similarity: String)
-  (implicit sc: SparkContext, conv: T => Document)
-  : FacetedLuceneRDD[T] = {
-    apply(sc.parallelize[T](elems.toSeq), indexAnalyzer, queryAnalyzer, similarity)
-  }
-
-  def apply[T : ClassTag]
-  (elems: Iterable[T])
-  (implicit sc: SparkContext, conv: T => Document)
-  : FacetedLuceneRDD[T] = {
-    apply(sc.parallelize[T](elems.toSeq))
-  }
-
-  /**
-   * Instantiate a FacetedLuceneRDD with DataFrame
-   *
-   * @param dataFrame Spark DataFrame
-   * @param indexAnalyzer Index Analyzer name
-   * @param queryAnalyzer Query Analyzer name
-   * @param similarity Lucene scoring similarity, i.e., BM25 or TF-IDF
-   * @return
-   */
-  def apply(dataFrame: DataFrame, indexAnalyzer: String, queryAnalyzer: String, similarity: String)
-  : FacetedLuceneRDD[Row] = {
-    apply(dataFrame.rdd, indexAnalyzer, queryAnalyzer, similarity: String)
-  }
-
-  def apply(dataFrame: DataFrame)
-  : FacetedLuceneRDD[Row] = {
-    apply(dataFrame.rdd, getOrElseEn(IndexAnalyzerConfigName), getOrElseEn(QueryAnalyzerConfigName),
-      getOrElseClassic())
-  }
-}
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/facets/package.scala b/src/main/scala/org/zouzias/spark/lucenerdd/facets/package.scala
deleted file mode 100644
index e2d6cf2a..00000000
--- a/src/main/scala/org/zouzias/spark/lucenerdd/facets/package.scala
+++ /dev/null
@@ -1,165 +0,0 @@
-/*
- * 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.
- */
-package org.zouzias.spark.lucenerdd
-
-import org.apache.lucene.document._
-import org.apache.lucene.facet.FacetField
-import org.apache.spark.sql.Row
-
-import scala.reflect.ClassTag
-
-/**
-  * Contains implicit conversion to [[org.apache.lucene.document.Document]]
-  * which prepares the index for faceted search as well.
-  */
-package object facets {
-
-  private val Stored = Field.Store.YES
-  private val DefaultFieldName = "_1"
-
-  /**
-    * Adds extra field on index with suffix [[FacetedLuceneRDD.FacetTextFieldSuffix]]
-    * This fiels is used on faceted queries
-    *
-    * @param doc Input document
-    * @param fieldName Field name
-    * @param fieldValue Field value to be indexed
-    */
-  private def addTextFacetField(doc: Document, fieldName: String, fieldValue: String): Unit = {
-    if ( fieldValue.nonEmpty) { // Issues with empty strings on facets
-      doc.add(new FacetField(s"${fieldName}${FacetedLuceneRDD.FacetTextFieldSuffix}",
-        fieldValue))
-    }
-  }
-
-  implicit def intToDocument(v: Int): Document = {
-    val doc = new Document
-    doc.add(new IntPoint(DefaultFieldName, v))
-    addTextFacetField(doc, DefaultFieldName, v.toString)
-    doc
-  }
-
-  implicit def longToDocument(v: Long): Document = {
-    val doc = new Document
-    doc.add(new LongPoint(DefaultFieldName, v))
-    addTextFacetField(doc, DefaultFieldName, v.toString)
-    doc
-  }
-
-  implicit def doubleToDocument(v: Double): Document = {
-    val doc = new Document
-    doc.add(new DoublePoint(DefaultFieldName, v))
-    addTextFacetField(doc, DefaultFieldName, v.toString)
-    doc
-  }
-
-  implicit def floatToDocument(v: Float): Document = {
-    val doc = new Document
-    doc.add(new FloatPoint(DefaultFieldName, v))
-    addTextFacetField(doc, DefaultFieldName, v.toString)
-    doc
-  }
-
-  implicit def stringToDocument(s: String): Document = {
-    val doc = new Document
-    doc.add(new TextField(DefaultFieldName, s, Stored))
-    addTextFacetField(doc, DefaultFieldName, s)
-    doc
-  }
-
-  private def tupleTypeToDocument[T: ClassTag](doc: Document, index: Int, s: T): Document = {
-    typeToDocument(doc, s"_${index}", s)
-  }
-
-  def typeToDocument[T: ClassTag](doc: Document, fName: String, s: T): Document = {
-    s match {
-      case x: String =>
-        doc.add(new TextField(fName, x, Stored))
-        addTextFacetField(doc, fName, x)
-      case x: Long =>
-        doc.add(new LongPoint(fName, x))
-        doc.add(new StoredField(fName, x))
-        doc.add(new NumericDocValuesField(s"${fName} ${FacetedLuceneRDD.FacetNumericFieldSuffix}",
-          x))
-      case x: Int =>
-        doc.add(new IntPoint(fName, x))
-        doc.add(new StoredField(fName, x))
-        doc.add(new NumericDocValuesField(s"${fName}${FacetedLuceneRDD.FacetNumericFieldSuffix}",
-          x.toLong))
-      case x: Float =>
-        doc.add(new FloatPoint(fName, x))
-        doc.add(new StoredField(fName, x))
-        doc.add(new FloatDocValuesField(s"${fName}${FacetedLuceneRDD.FacetNumericFieldSuffix}",
-          x))
-      case x: Double =>
-        doc.add(new DoublePoint(fName, x))
-        doc.add(new StoredField(fName, x))
-        doc.add(new DoubleDocValuesField(s"${fName}${FacetedLuceneRDD.FacetNumericFieldSuffix}",
-          x))
-    }
-    doc
-  }
-
-  implicit def iterablePrimitiveToDocument[T: ClassTag](iter: Iterable[T]): Document = {
-    val doc = new Document
-    iter.foreach( item => tupleTypeToDocument(doc, 1, item))
-    doc
-  }
-
-  implicit def mapToDocument[T: ClassTag](map: Map[String, T]): Document = {
-    val doc = new Document
-    map.foreach{ case (key, value) =>
-      typeToDocument(doc, key, value)
-    }
-    doc
-  }
-
-  /**
-   * Implicit conversion for all product types, such as case classes and Tuples
-   * @param s
-   * @tparam T
-   * @return
-   */
-  implicit def productTypeToDocument[T <: Product : ClassTag](s: T): Document = {
-    val doc = new Document
-
-    val fieldNames = s.getClass.getDeclaredFields.map(_.getName).toIterator
-    val fieldValues = s.productIterator
-    fieldValues.zip(fieldNames).foreach{ case (elem, fieldName) =>
-      typeToDocument(doc, fieldName, elem)
-    }
-
-    doc
-  }
-
-  /**
-   * Implicit conversion for Spark Row: used for DataFrame
-   * @param row
-   * @return
-   */
-  implicit def sparkRowToDocument(row: Row): Document = {
-    val doc = new Document
-
-    val fieldNames = row.schema.fieldNames
-    fieldNames.foreach{ case fieldName =>
-      val index = row.fieldIndex(fieldName)
-      typeToDocument(doc, fieldName, row.get(index))
-    }
-
-    doc
-  }
-}
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/models/SparkFacetResult.scala b/src/main/scala/org/zouzias/spark/lucenerdd/models/SparkFacetResult.scala
deleted file mode 100644
index 8d953b69..00000000
--- a/src/main/scala/org/zouzias/spark/lucenerdd/models/SparkFacetResult.scala
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.
- */
-package org.zouzias.spark.lucenerdd.models
-
-import org.apache.lucene.facet.FacetResult
-
-case class SparkFacetResult(facetName: String, facets: Map[String, Long]) {
-
-  /**
-   * Return facet counts sorted descending
-   * @return Sequence of (facet value, facet counts)
-   */
-  def sortedFacets(): Seq[(String, Long)] = {
-    facets.toSeq.sortBy[Long](x => -x._2)
-  }
-}
-
-
-object SparkFacetResult extends Serializable {
-
-  /**
-   * Convert [[org.apache.lucene.facet.FacetResult]]
-   * to [[org.zouzias.spark.lucenerdd.models.SparkFacetResult]]
-   *
-   * @param facetName name of facet
-   * @param facetResult input facet results
-   * @return
-   */
-  def apply(facetName: String, facetResult: FacetResult): SparkFacetResult = {
-    val facetResultOpt = Option(facetResult)
-    facetResultOpt match {
-    case Some(fctResult) =>
-      val map = fctResult.labelValues
-        .map(labelValue => (labelValue.label, labelValue.value.longValue()))
-        .toMap[String, Long]
-      SparkFacetResult(facetName, map)
-      case _ => SparkFacetResult(facetName, Map.empty[String, Long])
-    }
-  }
-}
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/package.scala b/src/main/scala/org/zouzias/spark/lucenerdd/package.scala
index eb0867a9..bc1f32b1 100644
--- a/src/main/scala/org/zouzias/spark/lucenerdd/package.scala
+++ b/src/main/scala/org/zouzias/spark/lucenerdd/package.scala
@@ -18,8 +18,10 @@ package org.zouzias.spark
 
 import org.apache.lucene.document._
 import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.ArrayType
 import org.zouzias.spark.lucenerdd.config.LuceneRDDConfigurable
 
+import collection.JavaConverters._
 import scala.reflect.ClassTag
 
 package object lucenerdd extends LuceneRDDConfigurable {
@@ -47,6 +49,14 @@ package object lucenerdd extends LuceneRDDConfigurable {
     }
   }
 
+  private def listPrimitiveToDocument[T: ClassTag](doc: Document,
+                                                   fieldName: String,
+                                                   iter: java.util.List[T])
+  : Document = {
+    iter.asScala.foreach( item => typeToDocument(doc, fieldName, item))
+    doc
+  }
+
   implicit def intToDocument(v: Int): Document = {
     val doc = new Document
     if (v != null) {
@@ -116,7 +126,6 @@ package object lucenerdd extends LuceneRDDConfigurable {
   }
 
   def typeToDocument[T: ClassTag](doc: Document, fieldName: String, s: T): Document = {
-
     s match {
       case x: String if x != null =>
         doc.add(new Field(fieldName, x,
@@ -135,7 +144,10 @@ package object lucenerdd extends LuceneRDDConfigurable {
       case x: Double if x != null =>
         doc.add(new DoublePoint(fieldName, x))
         doc.add(new StoredField(fieldName, x))
-      case _ => Unit
+      case null => Unit
+      case _ =>
+        throw new RuntimeException(s"Type ${s.getClass.getName} " +
+          s"on field ${fieldName} is not supported")
     }
     doc
   }
@@ -146,6 +158,12 @@ package object lucenerdd extends LuceneRDDConfigurable {
     doc
   }
 
+  implicit def arrayPrimitiveToDocument[T: ClassTag](iter: Array[T]): Document = {
+    val doc = new Document
+    iter.foreach( item => tupleTypeToDocument(doc, 1, item))
+    doc
+  }
+
   implicit def mapToDocument[T: ClassTag](map: Map[String, T]): Document = {
     val doc = new Document
     map.foreach{ case (key, value) =>
@@ -180,10 +198,17 @@ package object lucenerdd extends LuceneRDDConfigurable {
   implicit def sparkRowToDocument(row: Row): Document = {
     val doc = new Document
 
-    val fieldNames = row.schema.fieldNames
-    fieldNames.foreach{ case fieldName =>
+    row.schema.map(field => (field.name, field.dataType))
+      .foreach{ case (fieldName, dataType) =>
       val index = row.fieldIndex(fieldName)
-      typeToDocument(doc, fieldName, row.get(index))
+
+      // TODO: Handle org.apache.spark.sql.types.MapType and more
+      if (dataType.isInstanceOf[ArrayType]) {
+       listPrimitiveToDocument(doc, fieldName, row.getList(index))
+      }
+      else {
+        typeToDocument(doc, fieldName, row.get(index))
+      }
     }
 
     doc
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/partition/AbstractLuceneRDDPartition.scala b/src/main/scala/org/zouzias/spark/lucenerdd/partition/AbstractLuceneRDDPartition.scala
index 3a43c91d..fd0652de 100644
--- a/src/main/scala/org/zouzias/spark/lucenerdd/partition/AbstractLuceneRDDPartition.scala
+++ b/src/main/scala/org/zouzias/spark/lucenerdd/partition/AbstractLuceneRDDPartition.scala
@@ -18,7 +18,7 @@ package org.zouzias.spark.lucenerdd.partition
 
 import org.apache.lucene.search.{BooleanClause, Query}
 import org.zouzias.spark.lucenerdd.models.indexstats.IndexStatistics
-import org.zouzias.spark.lucenerdd.models.{SparkFacetResult, TermVectorEntry}
+import org.zouzias.spark.lucenerdd.models.TermVectorEntry
 import org.zouzias.spark.lucenerdd.response.LuceneRDDResponsePartition
 
 import scala.reflect.ClassTag
@@ -81,15 +81,6 @@ private[lucenerdd] abstract class AbstractLuceneRDDPartition[T] extends Serializ
   def queries(searchString: Iterable[String], topK: Int)
   : Iterable[(String, LuceneRDDResponsePartition)]
 
-  /**
-   * Generic Lucene faceted Query using QueryParser
-   * @param searchString Lucene query string, i.e., textField:hello*
-   * @param topK Number of facets to return
-   * @return
-   */
-  def facetQuery(searchString: String, facetField: String, topK: Int)
-  : SparkFacetResult
-
   /**
    * Term Query
    * @param fieldName Name of field
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/partition/LuceneRDDPartition.scala b/src/main/scala/org/zouzias/spark/lucenerdd/partition/LuceneRDDPartition.scala
index 3e0b4fe6..3c36a316 100644
--- a/src/main/scala/org/zouzias/spark/lucenerdd/partition/LuceneRDDPartition.scala
+++ b/src/main/scala/org/zouzias/spark/lucenerdd/partition/LuceneRDDPartition.scala
@@ -17,18 +17,18 @@
 package org.zouzias.spark.lucenerdd.partition
 
 import org.apache.lucene.analysis.Analyzer
+import org.apache.lucene.analysis.miscellaneous.PerFieldAnalyzerWrapper
 import org.apache.lucene.document._
-import org.apache.lucene.facet.taxonomy.directory.DirectoryTaxonomyReader
 import org.apache.lucene.index.{DirectoryReader, IndexReader}
 import org.apache.lucene.search._
 import org.joda.time.DateTime
-import org.zouzias.spark.lucenerdd.facets.FacetedLuceneRDD
+import scala.collection.JavaConverters._
 import org.zouzias.spark.lucenerdd.models.indexstats.{FieldStatistics, IndexStatistics}
-import org.zouzias.spark.lucenerdd.models.{SparkFacetResult, TermVectorEntry}
+import org.zouzias.spark.lucenerdd.models.TermVectorEntry
 import org.zouzias.spark.lucenerdd.query.{LuceneQueryHelpers, SimilarityConfigurable}
 import org.zouzias.spark.lucenerdd.response.LuceneRDDResponsePartition
-import org.zouzias.spark.lucenerdd.store.IndexWithTaxonomyWriter
 import org.zouzias.spark.lucenerdd.LuceneRDD
+import org.zouzias.spark.lucenerdd.store.IndexWritable
 
 import scala.reflect.{ClassTag, _}
 import scala.collection.mutable.ArrayBuffer
@@ -36,12 +36,19 @@ import scala.collection.mutable.ArrayBuffer
 /**
   * A partition of [[LuceneRDD]]
   *
-  * @param iter
-  * @param partitionId
-  * @param indexAnalyzerName
-  * @param queryAnalyzerName
-  * @param docConversion
-  * @param kTag
+  * Each LuceneRDD partition constructs a Lucene index by using [[Analyzer]]
+  * during indexing time and allowing to specify Lucene [[Analyzer]] during query
+  * time.
+  *
+  * @param iter Iterator of elements to be Lucene indexed
+  * @param partitionId Identifier of the RDD partition
+  * @param indexAnalyzerName Lucene Analyzer applied during index time
+  * @param queryAnalyzerName Lucene Analyzer applied during query time
+  * @param similarityName Lucene Similarity metric (BM25, Tf/idf)
+  * @param indexAnalyzerPerField Lucene Analyzer per field (indexing time)
+  * @param queryAnalyzerPerField Lucene Analyzer per field (query time)
+  * @param docConversion Conversion from T to a Lucene document
+  * @param kTag Class tag of type T
   * @tparam T the type associated with each entry in the set.
   */
 private[lucenerdd] class LuceneRDDPartition[T]
@@ -49,43 +56,56 @@ private[lucenerdd] class LuceneRDDPartition[T]
  private val partitionId: Int,
  private val indexAnalyzerName: String,
  private val queryAnalyzerName: String,
- private val similarityName: String)
+ private val similarityName: String,
+ private val indexAnalyzerPerField: Map[String, String],
+ private val queryAnalyzerPerField: Map[String, String])
 (implicit docConversion: T => Document,
  override implicit val kTag: ClassTag[T])
   extends AbstractLuceneRDDPartition[T]
-  with IndexWithTaxonomyWriter
+  with IndexWritable
   with SimilarityConfigurable {
 
-  logInfo(s"[partId=${partitionId}] Partition is created...")
+  logInfo(s"[partId=$partitionId] Partition is created...")
 
   override def indexAnalyzer(): Analyzer = getAnalyzer(Some(indexAnalyzerName))
 
+  override def indexPerFieldAnalyzer(): PerFieldAnalyzerWrapper = {
+    val analyzerPerField: Map[String, Analyzer] = indexAnalyzerPerField.mapValues(x =>
+      getAnalyzer(Some(x)))
+    new PerFieldAnalyzerWrapper(indexAnalyzer(), analyzerPerField.asJava)
+  }
+
   private val QueryAnalyzer: Analyzer = getAnalyzer(Some(queryAnalyzerName))
 
+  private def PerFieldQueryAnalyzer(): PerFieldAnalyzerWrapper = {
+    val analyzerPerField: Map[String, Analyzer] = queryAnalyzerPerField.mapValues(x =>
+      getAnalyzer(Some(x)))
+    new PerFieldAnalyzerWrapper(QueryAnalyzer, analyzerPerField.asJava)
+  }
+
   private val (iterOriginal, iterIndex) = iter.duplicate
 
   private val startTime = new DateTime(System.currentTimeMillis())
-  logInfo(s"[partId=${partitionId}]Indexing process initiated at ${startTime}...")
+  logInfo(s"[partId=$partitionId]Indexing process initiated at $startTime...")
   iterIndex.foreach { case elem =>
     // (implicitly) convert type T to Lucene document
     val doc = docConversion(elem)
-    indexWriter.addDocument(FacetsConfig.build(taxoWriter, doc))
+    indexWriter.addDocument(doc)
   }
   private val endTime = new DateTime(System.currentTimeMillis())
-  logInfo(s"[partId=${partitionId}]Indexing process completed at ${endTime}...")
-  logInfo(s"[partId=${partitionId}]Indexing process took ${(endTime.getMillis
+  logInfo(s"[partId=$partitionId]Indexing process completed at $endTime...")
+  logInfo(s"[partId=$partitionId]Indexing process took ${(endTime.getMillis
     - startTime.getMillis) / 1000} seconds...")
 
-  // Close the indexWriter and taxonomyWriter (for faceted search)
+  // Close the indexWriter
   closeAllWriters()
-  logDebug(s"[partId=${partitionId}]Closing index writers...")
+  logDebug(s"[partId=$partitionId]Closing index writers...")
 
-  logDebug(s"[partId=${partitionId}]Instantiating index/facet readers")
+  logDebug(s"[partId=$partitionId]Instantiating index readers")
   private val indexReader = DirectoryReader.open(IndexDir)
   private lazy val indexSearcher = initializeIndexSearcher(indexReader)
-  private val taxoReader = new DirectoryTaxonomyReader(TaxonomyDir)
-  logDebug(s"[partId=${partitionId}]Index readers instantiated successfully")
-  logInfo(s"[partId=${partitionId}]Indexed ${size} documents")
+  logDebug(s"[partId=$partitionId]Index readers instantiated successfully")
+  logInfo(s"[partId=$partitionId]Indexed $size documents")
 
   override def fields(): Set[String] = {
     LuceneQueryHelpers.fields(indexSearcher)
@@ -121,7 +141,8 @@ private[lucenerdd] class LuceneRDDPartition[T]
 
   override def filter(pred: T => Boolean): AbstractLuceneRDDPartition[T] =
     new LuceneRDDPartition(iterOriginal.filter(pred),
-      partitionId, indexAnalyzerName, queryAnalyzerName, similarityName)(docConversion, kTag)
+      partitionId, indexAnalyzerName, queryAnalyzerName, similarityName,
+      indexAnalyzerPerField, queryAnalyzerPerField)(docConversion, kTag)
 
   override def termQuery(fieldName: String, fieldText: String,
                          topK: Int = 1): LuceneRDDResponsePartition = {
@@ -132,7 +153,8 @@ private[lucenerdd] class LuceneRDDPartition[T]
 
   override def query(searchString: String,
                      topK: Int): LuceneRDDResponsePartition = {
-    val results = LuceneQueryHelpers.searchParser(indexSearcher, searchString, topK, QueryAnalyzer)
+    val results = LuceneQueryHelpers.searchParser(indexSearcher, searchString, topK,
+      PerFieldQueryAnalyzer())
 
     LuceneRDDResponsePartition(results.toIterator)
   }
@@ -174,15 +196,6 @@ private[lucenerdd] class LuceneRDDPartition[T]
     LuceneRDDResponsePartition(results.toIterator)
   }
 
-  override def facetQuery(searchString: String,
-                          facetField: String,
-                          topK: Int): SparkFacetResult = {
-    LuceneQueryHelpers.facetedTextSearch(indexSearcher, taxoReader, FacetsConfig,
-      searchString,
-      facetField + FacetedLuceneRDD.FacetTextFieldSuffix,
-      topK, QueryAnalyzer)
-  }
-
   override def moreLikeThis(fieldName: String, query: String,
                             minTermFreq: Int, minDocFreq: Int, topK: Int)
   : LuceneRDDResponsePartition = {
@@ -209,7 +222,7 @@ private[lucenerdd] class LuceneRDDPartition[T]
           case None => (docId.toString, partitionId)
         }
 
-        termsOpt.foreach { case terms =>
+        termsOpt.foreach { terms =>
 
           // Iterate over terms of each document
           val termsIter = terms.iterator()
@@ -241,23 +254,28 @@ object LuceneRDDPartition {
   /**
     * Constructor for [[LuceneRDDPartition]]
     *
-    * @param iter
-    * @param partitionId
-    * @param indexAnalyzer
-    * @param queryAnalyzer
-    * @param similarityName
-    * @param docConversion
-    * @tparam T
-    * @return
+    * @param iter Iterator of elements to be Lucene indexed
+    * @param partitionId Identifier of the RDD partition
+    * @param indexAnalyzerName Lucene Analyzer applied during index time
+    * @param queryAnalyzerName Lucene Analyzer applied during query time
+    * @param similarityName Lucene Similarity metric (BM25, Tf/idf)
+    * @param indexAnalyzerPerField Lucene Analyzer per field (indexing time), default empty
+    * @param queryAnalyzerPerField Lucene Analyzer per field (query time), default empty
+    * @param docConversion Convertion from T to a Lucene document
+    * @tparam T the type associated with each entry in the set.
+    * @return A partition of [[LuceneRDD]], type [[LuceneRDDPartition]]
     */
   def apply[T: ClassTag](iter: Iterator[T],
                          partitionId: Int,
-                         indexAnalyzer: String,
-                         queryAnalyzer: String,
-                         similarityName: String)
+                         indexAnalyzerName: String,
+                         queryAnalyzerName: String,
+                         similarityName: String,
+                         indexAnalyzerPerField: Map[String, String] = Map.empty,
+                         queryAnalyzerPerField: Map[String, String] = Map.empty)
                         (implicit docConversion: T => Document)
   : LuceneRDDPartition[T] = {
     new LuceneRDDPartition[T](iter, partitionId,
-      indexAnalyzer, queryAnalyzer, similarityName)(docConversion, classTag[T])
+      indexAnalyzerName, queryAnalyzerName, similarityName, indexAnalyzerPerField,
+      queryAnalyzerPerField)(docConversion, classTag[T])
   }
 }
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/query/LuceneQueryHelpers.scala b/src/main/scala/org/zouzias/spark/lucenerdd/query/LuceneQueryHelpers.scala
index 1b1a60ee..e957cdd9 100644
--- a/src/main/scala/org/zouzias/spark/lucenerdd/query/LuceneQueryHelpers.scala
+++ b/src/main/scala/org/zouzias/spark/lucenerdd/query/LuceneQueryHelpers.scala
@@ -19,17 +19,14 @@ package org.zouzias.spark.lucenerdd.query
 import java.io.StringReader
 
 import org.apache.lucene.analysis.Analyzer
+import org.apache.lucene.analysis.miscellaneous.PerFieldAnalyzerWrapper
 import org.apache.lucene.analysis.tokenattributes.CharTermAttribute
 import org.apache.lucene.document.Document
-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
-import org.zouzias.spark.lucenerdd.models.{SparkFacetResult, SparkScoreDoc}
+import org.zouzias.spark.lucenerdd.models.SparkScoreDoc
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable.ListBuffer
@@ -82,11 +79,12 @@ object LuceneQueryHelpers extends Serializable {
    * Parse a Query string
    *
    * @param searchString
-   * @param analyzer
+   * @param queryAnalyzerPerField Lucene query Analyzers per field
    * @return
    */
-  def parseQueryString(searchString: String, analyzer: Analyzer): Query = {
-    val queryParser = new QueryParser(QueryParserDefaultField, analyzer)
+  def parseQueryString(searchString: String, queryAnalyzerPerField: PerFieldAnalyzerWrapper)
+  : Query = {
+    val queryParser = new QueryParser(QueryParserDefaultField, queryAnalyzerPerField)
     queryParser.parse(searchString)
   }
 
@@ -96,15 +94,15 @@ object LuceneQueryHelpers extends Serializable {
    * @param indexSearcher Index searcher
    * @param searchString Lucene search query string
    * @param topK Number of documents to return
-   * @param analyzer Lucene Analyzer
+   * @param queryAnalyzerPerField Lucene Analyzer per field
    * @return
    */
   def searchParser(indexSearcher: IndexSearcher,
                    searchString: String,
                    topK: Int,
-                   analyzer: Analyzer)
+                   queryAnalyzerPerField: PerFieldAnalyzerWrapper)
   : Seq[SparkScoreDoc] = {
-    val q = parseQueryString(searchString, analyzer)
+    val q = parseQueryString(searchString, queryAnalyzerPerField)
     indexSearcher.search(q, topK).scoreDocs.map(SparkScoreDoc(indexSearcher, _))
   }
 
@@ -124,38 +122,6 @@ object LuceneQueryHelpers extends Serializable {
     indexSearcher.search(query, topK).scoreDocs.map(SparkScoreDoc(indexSearcher, _))
   }
 
-  /**
-   * Faceted search using [[SortedSetDocValuesFacetCounts]]
-   *
-   * @param indexSearcher Index searcher
-   * @param taxoReader taxonomy reader used for faceted search
-   * @param searchString Lucene search query string
-   * @param facetField Facet field name
-   * @param topK Number of returned documents
-   * @return
-   */
-  def facetedTextSearch(indexSearcher: IndexSearcher,
-                        taxoReader: TaxonomyReader,
-                        facetsConfig: FacetsConfig,
-                        searchString: String,
-                        facetField: String,
-                        topK: Int, analyzer: Analyzer): SparkFacetResult = {
-    // Prepare the query
-    val queryParser = new QueryParser(QueryParserDefaultField, analyzer)
-    val q: Query = queryParser.parse(searchString)
-
-    // Collect the facets
-    val fc = new FacetsCollector()
-    FacetsCollector.search(indexSearcher, q, topK, fc)
-    val facets = Option(new FastTaxonomyFacetCounts(taxoReader, facetsConfig, fc))
-
-    // Present the facets
-    facets match {
-      case Some(fcts) => SparkFacetResult(facetField, fcts.getTopChildren(topK, facetField))
-      case None => SparkFacetResultMonoid.zero(facetField)
-    }
-  }
-
   /**
    * Returns total number of lucene documents
    *
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/response/LuceneRDDResponse.scala b/src/main/scala/org/zouzias/spark/lucenerdd/response/LuceneRDDResponse.scala
index 31ff662e..4c195e74 100644
--- a/src/main/scala/org/zouzias/spark/lucenerdd/response/LuceneRDDResponse.scala
+++ b/src/main/scala/org/zouzias/spark/lucenerdd/response/LuceneRDDResponse.scala
@@ -76,9 +76,13 @@ private[lucenerdd] class LuceneRDDResponse
 
   override def collect(): Array[SparkScoreDoc] = {
     val sz = partitionsRDD.map(_.size).sum().toInt
-    val monoid = new TopKMonoid[SparkScoreDoc](sz)(ordering)
-    partitionsRDD.map(monoid.build(_))
-      .reduce(monoid.plus).items.toArray
+    if (sz > 0) {
+      val monoid = new TopKMonoid[SparkScoreDoc](sz)(ordering)
+      partitionsRDD.map(monoid.build(_))
+        .reduce(monoid.plus).items.toArray
+    } else {
+      Array.empty[SparkScoreDoc]
+    }
   }
 
   /**
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/ShapeLuceneRDDKryoRegistrator.scala b/src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/ShapeLuceneRDDKryoRegistrator.scala
index eb2b5269..420c91eb 100644
--- a/src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/ShapeLuceneRDDKryoRegistrator.scala
+++ b/src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/ShapeLuceneRDDKryoRegistrator.scala
@@ -22,7 +22,7 @@ import org.apache.spark.SparkConf
 import org.apache.spark.serializer.{KryoRegistrator, KryoSerializer}
 import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
 import org.apache.spark.sql.types._
-import org.zouzias.spark.lucenerdd.models.{SparkDoc, SparkFacetResult, SparkScoreDoc}
+import org.zouzias.spark.lucenerdd.models.{SparkDoc, SparkScoreDoc}
 import org.zouzias.spark.lucenerdd.spatial.shape.partition.ShapeLuceneRDDPartition
 
 
@@ -73,7 +73,6 @@ class ShapeLuceneRDDKryoRegistrator extends KryoRegistrator {
     kryo.register(classOf[scala.collection.immutable.Set$EmptySet$])
     kryo.register(classOf[scala.collection.immutable.Map[_, _]])
     kryo.register(classOf[Array[scala.collection.immutable.Map[_, _]]])
-    kryo.register(classOf[SparkFacetResult])
     kryo.register(classOf[SparkScoreDoc])
     kryo.register(classOf[TopK[_]])
 
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/partition/ShapeLuceneRDDPartition.scala b/src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/partition/ShapeLuceneRDDPartition.scala
index aaeac337..13e7cd12 100644
--- a/src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/partition/ShapeLuceneRDDPartition.scala
+++ b/src/main/scala/org/zouzias/spark/lucenerdd/spatial/shape/partition/ShapeLuceneRDDPartition.scala
@@ -17,11 +17,13 @@
 package org.zouzias.spark.lucenerdd.spatial.shape.partition
 
 import org.apache.lucene.analysis.Analyzer
+import org.apache.lucene.analysis.miscellaneous.PerFieldAnalyzerWrapper
 import org.apache.lucene.document.{Document, StoredField}
 import org.apache.lucene.index.DirectoryReader
 import org.apache.lucene.search.{IndexSearcher, ScoreDoc, Sort}
 import org.apache.lucene.spatial.query.{SpatialArgs, SpatialOperation}
 import org.joda.time.DateTime
+import scala.collection.JavaConverters._
 import org.locationtech.spatial4j.distance.DistanceUtils
 import org.locationtech.spatial4j.shape.Shape
 import org.zouzias.spark.lucenerdd.models.SparkScoreDoc
@@ -29,24 +31,32 @@ import org.zouzias.spark.lucenerdd.query.LuceneQueryHelpers
 import org.zouzias.spark.lucenerdd.response.LuceneRDDResponsePartition
 import org.zouzias.spark.lucenerdd.spatial.shape.ShapeLuceneRDD.PointType
 import org.zouzias.spark.lucenerdd.spatial.shape.strategies.SpatialStrategy
-import org.zouzias.spark.lucenerdd.store.IndexWithTaxonomyWriter
+import org.zouzias.spark.lucenerdd.store.IndexWritable
 
 import scala.reflect._
 
 private[shape] class ShapeLuceneRDDPartition[K, V]
   (private val iter: Iterator[(K, V)],
    private val indexAnalyzerName: String,
-   private val queryAnalyzerName: String)
+   private val queryAnalyzerName: String,
+   private val indexAnalyzerPerField: Map[String, String],
+   private val queryAnalyzerPerField: Map[String, String])
   (override implicit val kTag: ClassTag[K],
    override implicit val vTag: ClassTag[V])
   (implicit shapeConversion: K => Shape,
    docConversion: V => Document)
   extends AbstractShapeLuceneRDDPartition[K, V]
-    with IndexWithTaxonomyWriter
+    with IndexWritable
     with SpatialStrategy {
 
   override def indexAnalyzer(): Analyzer = getAnalyzer(Some(indexAnalyzerName))
 
+  override def indexPerFieldAnalyzer(): PerFieldAnalyzerWrapper = {
+    val analyzerPerField: Map[String, Analyzer] = indexAnalyzerPerField.mapValues(x =>
+      getAnalyzer(Some(x)))
+    new PerFieldAnalyzerWrapper(indexAnalyzer(), analyzerPerField.asJava)
+  }
+
   private val QueryAnalyzer: Analyzer = getAnalyzer(Some(queryAnalyzerName))
 
   private def decorateWithLocation(doc: Document, shapes: Iterable[Shape]): Document = {
@@ -73,13 +83,13 @@ private[shape] class ShapeLuceneRDDPartition[K, V]
     val doc = docConversion(value)
     val shape = shapeConversion(key)
     val docWithLocation = decorateWithLocation(doc, Seq(shape))
-    indexWriter.addDocument(FacetsConfig.build(taxoWriter, docWithLocation))
+    indexWriter.addDocument(docWithLocation)
   }
   private val endTime = new DateTime(System.currentTimeMillis())
   logInfo(s"Indexing process completed at ${endTime}...")
   logInfo(s"Indexing process took ${(endTime.getMillis - startTime.getMillis) / 1000} seconds...")
 
-  // Close the indexWriter and taxonomyWriter (for faceted search)
+  // Close the indexWriter
   closeAllWriters()
 
   private val indexReader = DirectoryReader.open(IndexDir)
@@ -140,7 +150,7 @@ private[shape] class ShapeLuceneRDDPartition[K, V]
     // false = ascending dist
     val distSort = new Sort(valueSource.getSortField(false)).rewrite(indexSearcher)
 
-    val query = LuceneQueryHelpers.parseQueryString(searchString, QueryAnalyzer)
+    val query = LuceneQueryHelpers.parseQueryString(searchString, indexPerFieldAnalyzer())
     val docs = indexSearcher.search(query, k, distSort)
 
     // Here we sorted on it, and the distance will get
@@ -219,10 +229,13 @@ object ShapeLuceneRDDPartition {
     */
   def apply[K: ClassTag, V: ClassTag](iter: Iterator[(K, V)],
                                       indexAnalyzer: String,
-                                      queryAnalyzer: String)
+                                      queryAnalyzer: String,
+                                      indexAnalyzerPerField: Map[String, String] = Map.empty,
+                                      queryAnalyzerPerField: Map[String, String] = Map.empty)
   (implicit shapeConv: K => Shape, docConv: V => Document)
   : ShapeLuceneRDDPartition[K, V] = {
     new ShapeLuceneRDDPartition[K, V](iter,
-      indexAnalyzer, queryAnalyzer)(classTag[K], classTag[V]) (shapeConv, docConv)
+      indexAnalyzer, queryAnalyzer,
+      indexAnalyzerPerField, queryAnalyzerPerField)(classTag[K], classTag[V]) (shapeConv, docConv)
   }
 }
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/store/IndexStorable.scala b/src/main/scala/org/zouzias/spark/lucenerdd/store/IndexStorable.scala
index c7e4f292..f8062b50 100644
--- a/src/main/scala/org/zouzias/spark/lucenerdd/store/IndexStorable.scala
+++ b/src/main/scala/org/zouzias/spark/lucenerdd/store/IndexStorable.scala
@@ -18,7 +18,6 @@ package org.zouzias.spark.lucenerdd.store
 
 import java.nio.file.{Files, Path}
 
-import org.apache.lucene.facet.FacetsConfig
 import org.apache.lucene.store._
 import org.zouzias.spark.lucenerdd.config.Configurable
 import org.zouzias.spark.lucenerdd.logging.Logging
@@ -35,8 +34,6 @@ trait IndexStorable extends Configurable
   with AutoCloseable
   with Logging {
 
-  protected lazy val FacetsConfig = new FacetsConfig()
-
   private val IndexStoreKey = "lucenerdd.index.store.mode"
 
   private val tmpJavaDir = System.getProperty("java.io.tmpdir")
@@ -46,14 +43,7 @@ trait IndexStorable extends Configurable
 
   private val indexDir = Files.createTempDirectory(indexDirName)
 
-  private val taxonomyDirName =
-    s"taxonomyDirectory-${System.currentTimeMillis()}.${Thread.currentThread().getId}"
-
-  private val taxonomyDir = Files.createTempDirectory(taxonomyDirName)
-
-  protected val IndexDir = storageMode(indexDir)
-
-  protected val TaxonomyDir = storageMode(taxonomyDir)
+  protected val IndexDir: Directory = storageMode(indexDir)
 
   /**
    * Select Lucene index storage implementation based on config
@@ -110,6 +100,5 @@ trait IndexStorable extends Configurable
 
   override def close(): Unit = {
     IndexDir.close()
-    TaxonomyDir.close()
   }
 }
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/store/IndexWithTaxonomyWriter.scala b/src/main/scala/org/zouzias/spark/lucenerdd/store/IndexWritable.scala
similarity index 79%
rename from src/main/scala/org/zouzias/spark/lucenerdd/store/IndexWithTaxonomyWriter.scala
rename to src/main/scala/org/zouzias/spark/lucenerdd/store/IndexWritable.scala
index 8e51f82c..151f19ef 100644
--- a/src/main/scala/org/zouzias/spark/lucenerdd/store/IndexWithTaxonomyWriter.scala
+++ b/src/main/scala/org/zouzias/spark/lucenerdd/store/IndexWritable.scala
@@ -17,29 +17,27 @@
 package org.zouzias.spark.lucenerdd.store
 
 import org.apache.lucene.analysis.Analyzer
-import org.apache.lucene.facet.taxonomy.directory.DirectoryTaxonomyWriter
+import org.apache.lucene.analysis.miscellaneous.PerFieldAnalyzerWrapper
 import org.apache.lucene.index.IndexWriterConfig.OpenMode
 import org.apache.lucene.index.{IndexWriter, IndexWriterConfig}
 import org.zouzias.spark.lucenerdd.analyzers.AnalyzerConfigurable
 
 /**
- * Index and Taxonomy Writer used for facet queries
+ * Index writer
  */
-trait IndexWithTaxonomyWriter extends IndexStorable
+trait IndexWritable extends IndexStorable
   with AnalyzerConfigurable {
 
   protected def indexAnalyzer(): Analyzer
 
+  protected def indexPerFieldAnalyzer(): PerFieldAnalyzerWrapper
+
   protected lazy val indexWriter = new IndexWriter(IndexDir,
-    new IndexWriterConfig(indexAnalyzer())
+    new IndexWriterConfig(indexPerFieldAnalyzer())
       .setOpenMode(OpenMode.CREATE))
 
-  protected lazy val taxoWriter = new DirectoryTaxonomyWriter(TaxonomyDir)
-
   protected def closeAllWriters(): Unit = {
     indexWriter.commit()
-    taxoWriter.commit()
-    taxoWriter.close()
     indexWriter.close()
   }
 }
diff --git a/src/main/scala/org/zouzias/spark/lucenerdd/testing/FavoriteCaseClass.scala b/src/main/scala/org/zouzias/spark/lucenerdd/testing/FavoriteCaseClass.scala
index a96ac946..dd8c6b84 100644
--- a/src/main/scala/org/zouzias/spark/lucenerdd/testing/FavoriteCaseClass.scala
+++ b/src/main/scala/org/zouzias/spark/lucenerdd/testing/FavoriteCaseClass.scala
@@ -17,3 +17,6 @@
 package org.zouzias.spark.lucenerdd.testing
 
 case class FavoriteCaseClass(name: String, age: Int, myLong: Long, myFloat: Float, email: String)
+
+case class MultivalueFavoriteCaseClass(names: Array[String], age: Int, myLong: Long,
+                                       myFloat: Float, email: String)
diff --git a/src/test/resources/reference.conf b/src/test/resources/reference.conf
index 17f72bbc..a76ca011 100644
--- a/src/test/resources/reference.conf
+++ b/src/test/resources/reference.conf
@@ -59,9 +59,6 @@ lucenerdd {
   // Default value of number of returned results
   query.topk.default = 10
 
-  // Default value of number of faceted results
-  query.facets.number.default = 10
-
   // Spatial related configurations used by ShapeLuceneRDD
   spatial {
     prefixtree {
diff --git a/src/test/scala/org/zouzias/spark/lucenerdd/LucenePrimitiveTypesSpec.scala b/src/test/scala/org/zouzias/spark/lucenerdd/LucenePrimitiveTypesSpec.scala
index a0d735f7..41b422d4 100644
--- a/src/test/scala/org/zouzias/spark/lucenerdd/LucenePrimitiveTypesSpec.scala
+++ b/src/test/scala/org/zouzias/spark/lucenerdd/LucenePrimitiveTypesSpec.scala
@@ -40,14 +40,21 @@ class LucenePrimitiveTypesSpec extends FlatSpec with Matchers
   }
 
   /**
-   * Do not work with facets (multi-valued issue)
-
   "LuceneRDD" should "work with RDD[List[String]]" in {
     val array = Array(List("aaa", "aaa2"), List("bbb", "bbb2"),
       List("ccc", "ccc2"), List("ddd"), List("eee"))
     val rdd = sc.parallelize(array)
     luceneRDD = LuceneRDD(rdd)
-    luceneRDD.count should be (array.size)
+    luceneRDD.count should be (array.length)
+  }
+  */
+
+  "LuceneRDD" should "work with RDD[Array[String]]" in {
+    val array = Array(Array("aaa", "aaa2"), Array("bbb", "bbb2"),
+      Array("ccc", "ccc2"), Array("ddd"), Array("eee"))
+    val rdd = sc.parallelize(array)
+    luceneRDD = LuceneRDD(rdd)
+    luceneRDD.count should be (array.length)
   }
 
   "LuceneRDD" should "work with RDD[Set[String]]" in {
@@ -55,16 +62,14 @@ class LucenePrimitiveTypesSpec extends FlatSpec with Matchers
       Set("ccc", "ccc2"), Set("ddd"), Set("eee"))
     val rdd = sc.parallelize(array)
     luceneRDD = LuceneRDD(rdd)
-    luceneRDD.count should be (array.size)
+    luceneRDD.count should be (array.length)
   }
 
-   */
-
   "LuceneRDD" should "work with RDD[String]" in {
     val array = Array("aaa", "bbb", "ccc", "ddd", "eee")
     val rdd = sc.parallelize(array)
     luceneRDD = LuceneRDD(rdd)
-    luceneRDD.count should be (array.size)
+    luceneRDD.count should be (array.length)
   }
 
   "LuceneRDD" should "work with RDD[Int]" in {
@@ -109,7 +114,7 @@ class LucenePrimitiveTypesSpec extends FlatSpec with Matchers
     val array = Array("aaa", null, "ccc", null, "eee")
     val rdd = sc.parallelize(array)
     luceneRDD = LuceneRDD(rdd)
-    luceneRDD.count should be (array.size)
+    luceneRDD.count should be (array.length)
   }
 
 }
\ No newline at end of file
diff --git a/src/test/scala/org/zouzias/spark/lucenerdd/LuceneRDDCustomCaseClassImplicitsSpec.scala b/src/test/scala/org/zouzias/spark/lucenerdd/LuceneRDDCustomCaseClassImplicitsSpec.scala
index 4c55bea7..e8e53bb4 100644
--- a/src/test/scala/org/zouzias/spark/lucenerdd/LuceneRDDCustomCaseClassImplicitsSpec.scala
+++ b/src/test/scala/org/zouzias/spark/lucenerdd/LuceneRDDCustomCaseClassImplicitsSpec.scala
@@ -38,7 +38,7 @@ class LuceneRDDCustomCaseClassImplicitsSpec extends FlatSpec
     set("spark.ui.enabled", "false").
     set("spark.app.id", appID))
 
-  val elem = Array("fear", "death", "water", "fire", "house")
+  val elem: Array[Person] = Array("fear", "death", "water", "fire", "house")
     .zipWithIndex.map{ case (str, index) => Person(str, index, s"${str}@gmail.com")}
 
   "LuceneRDD(case class).count" should "handle nulls properly" in {
diff --git a/src/test/scala/org/zouzias/spark/lucenerdd/LuceneRDDDataFrameImplicitsSpec.scala b/src/test/scala/org/zouzias/spark/lucenerdd/LuceneRDDDataFrameImplicitsSpec.scala
index e5e7fdbc..415a2663 100644
--- a/src/test/scala/org/zouzias/spark/lucenerdd/LuceneRDDDataFrameImplicitsSpec.scala
+++ b/src/test/scala/org/zouzias/spark/lucenerdd/LuceneRDDDataFrameImplicitsSpec.scala
@@ -20,7 +20,7 @@ import com.holdenkarau.spark.testing.SharedSparkContext
 import org.apache.spark.SparkConf
 import org.apache.spark.sql.SparkSession
 import org.scalatest.{BeforeAndAfterEach, FlatSpec, Matchers}
-import org.zouzias.spark.lucenerdd.testing.FavoriteCaseClass
+import org.zouzias.spark.lucenerdd.testing.{FavoriteCaseClass, MultivalueFavoriteCaseClass}
 
 class LuceneRDDDataFrameImplicitsSpec extends FlatSpec
   with Matchers
@@ -45,6 +45,18 @@ class LuceneRDDDataFrameImplicitsSpec extends FlatSpec
     .zipWithIndex.map{ case (str, index) =>
     FavoriteCaseClass(str, index, 10L, 12.3F, s"${str}@gmail.com")}
 
+  val multiValuesElems = Array("fear", "death", "water", "fire", "house")
+    .zipWithIndex.map{ case (str, index) =>
+    MultivalueFavoriteCaseClass(Array(str, str.reverse), index, 10L, 12.3F, s"${str}@gmail.com")}
+
+  "LuceneRDD(MultivalueFavoriteCaseClass).count" should "return correct number of elements" in {
+    val rdd = sc.parallelize(multiValuesElems)
+    val spark = SparkSession.builder().getOrCreate()
+    import spark.implicits._
+    val df = rdd.toDF()
+    luceneRDD = LuceneRDD(df)
+    luceneRDD.count should equal (elem.length)
+  }
 
   "LuceneRDD(case class).count" should "return correct number of elements" in {
     val rdd = sc.parallelize(elem)
@@ -52,7 +64,7 @@ class LuceneRDDDataFrameImplicitsSpec extends FlatSpec
     import spark.implicits._
     val df = rdd.toDF()
     luceneRDD = LuceneRDD(df)
-    luceneRDD.count should equal (elem.size)
+    luceneRDD.count should equal (elem.length)
   }
 
   "LuceneRDD(case class).fields" should "return all fields" in {
@@ -70,6 +82,21 @@ class LuceneRDDDataFrameImplicitsSpec extends FlatSpec
     luceneRDD.fields().contains("email") should equal(true)
   }
 
+  "LuceneRDD(MultivalueFavoriteCaseClass).fields" should "return all fields" in {
+    val rdd = sc.parallelize(multiValuesElems)
+    val spark = SparkSession.builder().getOrCreate()
+    import spark.implicits._
+    val df = rdd.toDF()
+    luceneRDD = LuceneRDD(df)
+
+    luceneRDD.fields().size should equal(5)
+    luceneRDD.fields().contains("names") should equal(true)
+    luceneRDD.fields().contains("age") should equal(true)
+    luceneRDD.fields().contains("myLong") should equal(true)
+    luceneRDD.fields().contains("myFloat") should equal(true)
+    luceneRDD.fields().contains("email") should equal(true)
+  }
+
   "LuceneRDD(case class).termQuery" should "correctly search with TermQueries" in {
     val rdd = sc.parallelize(elem)
     val spark = SparkSession.builder().getOrCreate()
@@ -80,4 +107,15 @@ class LuceneRDDDataFrameImplicitsSpec extends FlatSpec
     val results = luceneRDD.termQuery("name", "water")
     results.count should equal(1)
   }
+
+  "LuceneRDD(MultivalueFavoriteCaseClass).termQuery" should "correctly search with TermQueries" in {
+    val rdd = sc.parallelize(multiValuesElems)
+    val spark = SparkSession.builder().getOrCreate()
+    import spark.implicits._
+    val df = rdd.toDF()
+    luceneRDD = LuceneRDD(df)
+
+    val results = luceneRDD.termQuery("names", "retaw")
+    results.count should equal(1)
+  }
 }
diff --git a/src/test/scala/org/zouzias/spark/lucenerdd/analyzers/AnalyzersConfigurableSpec.scala b/src/test/scala/org/zouzias/spark/lucenerdd/analyzers/AnalyzersConfigurableSpec.scala
index df933152..bbbd6986 100644
--- a/src/test/scala/org/zouzias/spark/lucenerdd/analyzers/AnalyzersConfigurableSpec.scala
+++ b/src/test/scala/org/zouzias/spark/lucenerdd/analyzers/AnalyzersConfigurableSpec.scala
@@ -17,6 +17,8 @@
 package org.zouzias.spark.lucenerdd.analyzers
 
 import org.apache.lucene.analysis.en.EnglishAnalyzer
+import org.apache.lucene.analysis.el.GreekAnalyzer
+import org.apache.lucene.analysis.de.GermanAnalyzer
 import org.scalatest.{BeforeAndAfterEach, FlatSpec, Matchers}
 
 class AnalyzersConfigurableSpec extends FlatSpec with Matchers
@@ -25,6 +27,21 @@ class AnalyzersConfigurableSpec extends FlatSpec with Matchers
 
   "AnalyzersConfigurable.getAnalyzer" should "return english analyzer with 'en' input" in {
     val englishAnalyzer = getAnalyzer(Some("en"))
+    englishAnalyzer shouldNot equal(null)
     englishAnalyzer.isInstanceOf[EnglishAnalyzer] should equal(true)
   }
+
+  "AnalyzersConfigurable.getAnalyzer" should
+    "return custom test analyzer with 'org.apache.lucene.analysis.el.GreekAnalyzer'" in {
+    val greekAnalyzer = getAnalyzer(Some("org.apache.lucene.analysis.el.GreekAnalyzer"))
+    greekAnalyzer shouldNot equal(null)
+    greekAnalyzer.isInstanceOf[GreekAnalyzer] should equal(true)
+  }
+
+  "AnalyzersConfigurable.getAnalyzer" should
+    "return custom test analyzer with 'org.apache.lucene.analysis.de.GermanAnalyzer'" in {
+    val deutschAnalyzer = getAnalyzer(Some("org.apache.lucene.analysis.de.GermanAnalyzer"))
+    deutschAnalyzer shouldNot equal(null)
+    deutschAnalyzer.isInstanceOf[GermanAnalyzer] should equal(true)
+  }
 }
diff --git a/src/test/scala/org/zouzias/spark/lucenerdd/facets/FacetedLuceneRDDFacetSpec.scala b/src/test/scala/org/zouzias/spark/lucenerdd/facets/FacetedLuceneRDDFacetSpec.scala
deleted file mode 100644
index b26fe3f5..00000000
--- a/src/test/scala/org/zouzias/spark/lucenerdd/facets/FacetedLuceneRDDFacetSpec.scala
+++ /dev/null
@@ -1,178 +0,0 @@
-/*
- * 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.
- */
-package org.zouzias.spark.lucenerdd.facets
-
-import com.holdenkarau.spark.testing.SharedSparkContext
-import org.apache.spark.SparkConf
-import org.scalatest.{BeforeAndAfterEach, FlatSpec, Matchers}
-import org.zouzias.spark.lucenerdd.{LuceneRDD, LuceneRDDKryoRegistrator}
-
-class FacetedLuceneRDDFacetSpec extends FlatSpec
-  with Matchers
-  with BeforeAndAfterEach
-  with SharedSparkContext {
-
-  override val conf = LuceneRDDKryoRegistrator.registerKryoClasses(new SparkConf().
-    setMaster("local[*]").
-    setAppName("test").
-    set("spark.ui.enabled", "false").
-    set("spark.app.id", appID))
-
-  // Check if sequence is sorted in descending order
-  def sortedDesc(seq : Seq[Long]) : Boolean = {
-    if (seq.isEmpty) true else seq.zip(seq.tail).forall(x => x._1 >= x._2)
-  }
-
-  "FacetedLuceneRDD.facetQuery" should "compute facets correctly" in {
-    val words = Array("aaa", "aaa", "aaa", "aaa", "bb", "bb", "bb", "cc", "cc")
-    val rdd = sc.parallelize(words)
-    val luceneRDD = FacetedLuceneRDD(rdd)
-    val facetResults = luceneRDD.facetQuery("*:*", "_1")._2
-
-    facetResults.facets.size should equal (3)
-    facetResults.facets.contains("aaa") should equal (true)
-    facetResults.facets.get("aaa")
-      .foreach(value => value should equal (4))
-
-    luceneRDD.close()
-  }
-
-  "FacetedLuceneRDD.facetQuery" should "compute facets correctly with ints" in {
-    val words = Array(10, 10, 10, 10, 22, 22, 22, 33, 33)
-    val rdd = sc.parallelize(words)
-    val luceneRDD = FacetedLuceneRDD(rdd)
-    val facetResults = luceneRDD.facetQuery("*:*", "_1")._2
-
-    facetResults.facets.size should equal (3)
-    facetResults.facets.contains("10") should equal (true)
-    facetResults.facets.contains("22") should equal (true)
-    facetResults.facets.contains("33") should equal (true)
-    facetResults.facets.get("10").foreach(value => value should equal (4))
-    facetResults.facets.get("33").foreach(value => value should equal (2))
-
-    luceneRDD.close()
-  }
-
-  "FacetedLuceneRDD.facetQuery" should "compute facets correctly with doubles" in {
-    val words = Array(10.5D, 10.5D, 10.5D, 10.5D, 22.2D, 22.2D, 22.2D, 33.2D, 33.2D)
-    val rdd = sc.parallelize(words)
-    val luceneRDD = FacetedLuceneRDD(rdd)
-    val facetResults = luceneRDD.facetQuery("*:*", "_1")._2
-
-    facetResults.facets.size should equal (3)
-    facetResults.facets.contains("10.5") should equal (true)
-    facetResults.facets.contains("22.2") should equal (true)
-    facetResults.facets.contains("33.2") should equal (true)
-    facetResults.facets.get("10.5").foreach(value => value should equal (4))
-    facetResults.facets.get("33.2").foreach(value => value should equal (2))
-
-    luceneRDD.close()
-  }
-
-  "FacetedLuceneRDD.facetQueries" should "compute facets correctly" in {
-    val words = Array("aaa", "aaa", "aaa", "aaa", "bb", "bb", "bb", "cc", "cc")
-    val rdd = sc.parallelize(words)
-    val luceneRDD = FacetedLuceneRDD(rdd)
-
-    val facetResults = luceneRDD.facetQueries("*:*", Seq("_1"))._2
-
-    facetResults.contains("_1") should equal(true)
-    facetResults.foreach(_._2.facets.size should equal (3))
-    facetResults.foreach(_._2.facets.contains("aaa") should equal (true))
-    facetResults.foreach(_._2.facets.get("aaa").foreach(value => value should equal (4)))
-
-    luceneRDD.close()
-  }
-
-  "FacetedLuceneRDD.sortedFacets" should "return facets sorted by decreasing order" in {
-    val words = Array("aaa", "aaa", "aaa", "aaa", "bb", "bb", "bb", "cc", "cc")
-    val rdd = sc.parallelize(words)
-    val luceneRDD = FacetedLuceneRDD(rdd)
-
-    val sortedFacetCounts = luceneRDD.facetQuery("*:*", "_1")._2.sortedFacets().map(_._2)
-    sortedDesc(sortedFacetCounts) should equal(true)
-
-    luceneRDD.close()
-  }
-
-  "FacetedLuceneRDD.facetQuery" should "compute facets with prefix search" in {
-    val words = Array("aaa", "aaa", "aaa", "aaa", "bb", "bb", "bb", "cc", "cc")
-    val rdd = sc.parallelize(words)
-    val luceneRDD = FacetedLuceneRDD(rdd)
-    val results = luceneRDD.facetQuery("_1:aa*", "_1")
-    val facetResults = results._2
-
-    facetResults.facets.size should equal (1)
-    facetResults.facets.contains("aaa") should equal (true)
-    facetResults.facets.get("aaa")
-      .foreach(value => value should equal (4))
-
-    luceneRDD.close()
-  }
-
-  "FacetedLuceneRDD.facetQuery" should "compute facets with term search" in {
-    val words = Array("aaa", "aaa", "aaa", "aaa", "aaaa", "bb", "bb", "bb", "cc", "cc")
-    val rdd = sc.parallelize(words)
-    val luceneRDD = FacetedLuceneRDD(rdd)
-    val results = luceneRDD.facetQuery("_1:aaa", "_1")
-    val facetResults = results._2
-
-    facetResults.facets.size should equal (1)
-    facetResults.facets.contains("aaa") should equal (true)
-    facetResults.facets.contains("bb") should equal (false)
-    facetResults.facets.contains("cc") should equal (false)
-    facetResults.facets.get("aaa") should equal (Some(4))
-
-    val resultsB = luceneRDD.facetQuery("_1:bb", "_1")
-    val facetResultsB = resultsB._2
-
-    facetResultsB.facets.contains("bb") should equal (true)
-    facetResultsB.facets.get("bb") should equal (Some(3))
-
-    luceneRDD.close()
-  }
-
-  "FacetedLuceneRDD.facetQuery" should "compute facets with term search in Tuple2" in {
-    val words = Array(("aaa", "aaa1"), ("aaa", "aaa2"), ("aaa", "aaa3"), ("aaa", "aaa3"),
-      ("aaaa", "aaa3"), ("bb", "cc1"), ("bb", "cc1"), ("bb", "cc1"), ("cc", "cc2"), ("cc", "cc2"))
-    val rdd = sc.parallelize(words)
-    val luceneRDD = FacetedLuceneRDD(rdd)
-    val results = luceneRDD.facetQuery("_1:aaa", "_2")
-    val facetResults = results._2
-
-    facetResults.facets.size should equal (3)
-    facetResults.facets.contains("aaa1") should equal (true)
-    facetResults.facets.contains("aaa2") should equal (true)
-    facetResults.facets.contains("aaa3") should equal (true)
-    facetResults.facets.get("aaa1") should equal (Some(1))
-    facetResults.facets.get("aaa2") should equal (Some(1))
-    facetResults.facets.get("aaa3") should equal (Some(2))
-
-    luceneRDD.close()
-  }
-
-  "FacetedLuceneRDD.version" should "return project sbt build information" in {
-    val map = LuceneRDD.version()
-    map.contains("name") should equal(true)
-    map.contains("builtAtMillis") should equal(true)
-    map.contains("scalaVersion") should equal(true)
-    map.contains("version") should equal(true)
-    map.contains("sbtVersion") should equal(true)
-    map.contains("builtAtString") should equal(true)
-  }
-
-}
diff --git a/src/test/scala/org/zouzias/spark/lucenerdd/facets/FacetedLuceneRDDImplicitsSpec.scala b/src/test/scala/org/zouzias/spark/lucenerdd/facets/FacetedLuceneRDDImplicitsSpec.scala
deleted file mode 100644
index e4348292..00000000
--- a/src/test/scala/org/zouzias/spark/lucenerdd/facets/FacetedLuceneRDDImplicitsSpec.scala
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * 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.
- */
-package org.zouzias.spark.lucenerdd.facets
-
-import com.holdenkarau.spark.testing.SharedSparkContext
-import org.apache.spark.SparkConf
-import org.apache.spark.sql.SparkSession
-import org.scalatest.{BeforeAndAfterEach, FlatSpec, Matchers}
-import org.zouzias.spark.lucenerdd.testing.FavoriteCaseClass
-import org.zouzias.spark.lucenerdd.{LuceneRDD, LuceneRDDKryoRegistrator}
-
-class FacetedLuceneRDDImplicitsSpec  extends FlatSpec
-  with Matchers
-  with BeforeAndAfterEach
-  with SharedSparkContext {
-
-  var luceneRDD: LuceneRDD[_] = _
-
-
-  override val conf = LuceneRDDKryoRegistrator.registerKryoClasses(new SparkConf().
-    setMaster("local[*]").
-    setAppName("test").
-    set("spark.ui.enabled", "false").
-    set("spark.app.id", appID))
-
-  override def afterEach() {
-    luceneRDD.close()
-  }
-
-
-  val elem = Array("fear", "death", "water", "fire", "house")
-    .zipWithIndex.map{ case (str, index) =>
-    FavoriteCaseClass(str, index, 10L, 12.3F, s"${str}@gmail.com")}
-
-
-  "FacetedLuceneRDD(case class).count" should "return correct number of elements" in {
-    val rdd = sc.parallelize(elem)
-    val spark = SparkSession.builder().getOrCreate()
-    import spark.implicits._
-    val df = rdd.toDF()
-    luceneRDD = FacetedLuceneRDD(df)
-    luceneRDD.count should equal (elem.size)
-  }
-
-  "FacetedLuceneRDD(case class).fields" should "return all fields" in {
-    val rdd = sc.parallelize(elem)
-    val spark = SparkSession.builder().getOrCreate()
-    import spark.implicits._
-    val df = rdd.toDF()
-    luceneRDD = FacetedLuceneRDD(df)
-
-    luceneRDD.fields().size should equal(5)
-    luceneRDD.fields().contains("name") should equal(true)
-    luceneRDD.fields().contains("age") should equal(true)
-    luceneRDD.fields().contains("myLong") should equal(true)
-    luceneRDD.fields().contains("myFloat") should equal(true)
-    luceneRDD.fields().contains("email") should equal(true)
-  }
-
-  "FacetedLuceneRDD(case class).termQuery" should "correctly search with TermQueries" in {
-    val rdd = sc.parallelize(elem)
-    val spark = SparkSession.builder().getOrCreate()
-    import spark.implicits._
-    val df = rdd.toDF()
-    luceneRDD = FacetedLuceneRDD(df)
-
-    val results = luceneRDD.termQuery("name", "water")
-    results.count() should equal(1)
-  }
-}
diff --git a/src/test/scala/org/zouzias/spark/lucenerdd/query/LuceneQueryHelpersSpec.scala b/src/test/scala/org/zouzias/spark/lucenerdd/query/LuceneQueryHelpersSpec.scala
index d9fbf85c..c4be755c 100644
--- a/src/test/scala/org/zouzias/spark/lucenerdd/query/LuceneQueryHelpersSpec.scala
+++ b/src/test/scala/org/zouzias/spark/lucenerdd/query/LuceneQueryHelpersSpec.scala
@@ -17,54 +17,55 @@
 package org.zouzias.spark.lucenerdd.query
 
 import org.apache.lucene.analysis.Analyzer
+import org.apache.lucene.analysis.miscellaneous.PerFieldAnalyzerWrapper
 import org.apache.lucene.document.Field.Store
 import org.apache.lucene.document._
-import org.apache.lucene.facet.FacetField
-import org.apache.lucene.facet.taxonomy.directory.DirectoryTaxonomyReader
+import scala.collection.JavaConverters._
 import org.apache.lucene.index.DirectoryReader
 import org.apache.lucene.search.IndexSearcher
 import org.scalatest.{BeforeAndAfterEach, FlatSpec, Matchers}
-import org.zouzias.spark.lucenerdd.facets.FacetedLuceneRDD
-import org.zouzias.spark.lucenerdd.store.IndexWithTaxonomyWriter
+import org.zouzias.spark.lucenerdd.store.IndexWritable
 
 import scala.io.Source
 
 class LuceneQueryHelpersSpec extends FlatSpec
-  with IndexWithTaxonomyWriter
+  with IndexWritable
   with Matchers
   with BeforeAndAfterEach {
 
   // Load cities
-  val countries = Source.fromFile("src/test/resources/countries.txt").getLines()
+  val countries: Seq[String] = Source.fromFile("src/test/resources/countries.txt").getLines()
     .map(_.toLowerCase()).toSeq
 
-  private val MaxFacetValue: Int = 10
+  val indexAnalyzerPerField: Map[String, String] = Map("name"
+    -> "org.apache.lucene.en.EnglishAnalyzer")
 
-  override def indexAnalyzer: Analyzer = getAnalyzer(Some("en"))
+  override def indexAnalyzer(): Analyzer = getAnalyzer(Some("en"))
+
+  override def indexPerFieldAnalyzer(): PerFieldAnalyzerWrapper = {
+    val analyzerPerField: Map[String, Analyzer] = indexAnalyzerPerField
+      .mapValues(x => getAnalyzer(Some(x)))
+    new PerFieldAnalyzerWrapper(indexAnalyzer(), analyzerPerField.asJava)
+  }
 
   countries.zipWithIndex.foreach { case (elem, index) =>
-    val doc = convertToDoc(index % MaxFacetValue, elem)
-    indexWriter.addDocument(FacetsConfig.build(taxoWriter, doc))
+    val doc = convertToDoc(index, elem)
+    indexWriter.addDocument(doc)
   }
 
   indexWriter.commit()
-  taxoWriter.close()
   indexWriter.close()
 
   private val indexReader = DirectoryReader.open(IndexDir)
   private val indexSearcher = new IndexSearcher(indexReader)
-  private lazy val taxoReader = new DirectoryTaxonomyReader(TaxonomyDir)
 
 
-  private lazy val TestFacetName = s"_2${FacetedLuceneRDD.FacetTextFieldSuffix}"
 
   def convertToDoc(pos: Int, text: String): Document = {
     val doc = new Document()
     doc.add(new StringField("_1", text, Store.YES))
-    doc.add(new FacetField(s"_1${FacetedLuceneRDD.FacetTextFieldSuffix}", text))
     doc.add(new IntPoint("_2", pos))
     doc.add(new StoredField("_2", pos))
-    doc.add(new FacetField(TestFacetName, pos.toString))
     doc
   }
 
@@ -76,14 +77,6 @@ class LuceneQueryHelpersSpec extends FlatSpec
     LuceneQueryHelpers.totalDocs(indexSearcher) should equal (countries.size)
   }
 
-  "LuceneQueryHelpers.facetedTextSearch" should "return correct facet counts" in {
-    val facets = LuceneQueryHelpers.facetedTextSearch(indexSearcher, taxoReader,
-      FacetsConfig, "*:*", TestFacetName, 100, indexAnalyzer)
-
-    facets.facetName should equal(TestFacetName)
-    facets.facets.size should equal(MaxFacetValue)
-  }
-
   "LuceneQueryHelpers.termQuery" should "return correct documents" in {
     val greece = "greece"
     val topDocs = LuceneQueryHelpers.termQuery(indexSearcher, "_1", greece, 100)
@@ -101,4 +94,5 @@ class LuceneQueryHelpersSpec extends FlatSpec
     topDocs.forall(doc => doc.doc.textField("_1").exists(x =>
       x.toString().toLowerCase().contains(prefix))) should equal(true)
   }
+
 }
diff --git a/src/test/scala/org/zouzias/spark/lucenerdd/response/LuceneRDDResponseSpec.scala b/src/test/scala/org/zouzias/spark/lucenerdd/response/LuceneRDDResponseSpec.scala
index cbcd9068..02ac27ca 100644
--- a/src/test/scala/org/zouzias/spark/lucenerdd/response/LuceneRDDResponseSpec.scala
+++ b/src/test/scala/org/zouzias/spark/lucenerdd/response/LuceneRDDResponseSpec.scala
@@ -90,4 +90,13 @@ class LuceneRDDResponseSpec extends FlatSpec with Matchers
       equal(org.apache.spark.sql.types.FloatType)
     schema.fields(schema.fieldIndex("email")).dataType should equal(StringType)
   }
+
+
+  "LuceneRDDResponseSpec.collect()" should "work when no results are found" in {
+    val array = Array("aaa", "bbb", "ccc", "ddd", "eee")
+    val rdd = sc.parallelize(array)
+    luceneRDD = LuceneRDD(rdd)
+    val result = luceneRDD.query("fff", 10)
+    result.collect().length should be (0)
+  }
 }