package com.microsoft.partnercatalyst.fortis.spark.transforms.topic

import com.microsoft.partnercatalyst.fortis.spark.transforms.Tag
import org.apache.commons.collections4.trie.PatriciaTrie

import scala.collection.mutable.ListBuffer

@SerialVersionUID(100L)
class KeywordExtractor(keywords: Seq[String]) extends Serializable {
  @transient private lazy val wordTokenizer = """\b""".r
  @transient private lazy val keywordTrie = initializeTrie(keywords)

  def extractKeywords(text: String): List[Tag] = {

    def findMatches(segment: Seq[String]): Iterable[String] = {
      val sb = new StringBuilder()
      val result = ListBuffer[String]()

      val it = segment.iterator
      var prefix = ""
      while (it.hasNext && !keywordTrie.prefixMap(prefix).isEmpty) {
        prefix = sb.append(it.next()).mkString

        Option(keywordTrie.get(prefix)).foreach(result.append(_))
      }

      result
    }

    val tokens = wordTokenizer.split(text.toLowerCase).toSeq
    tokens.tails.flatMap(findMatches(_).map(Tag(_, 1))).toList
  }

  private def initializeTrie(keywords: Seq[String]): PatriciaTrie[String] = {
    val trie = new PatriciaTrie[String]()
    keywords.foreach(k => trie.put(k.toLowerCase, k))

    trie
  }
}
package com.microsoft.partnercatalyst.fortis.spark.transforms.topic

import com.microsoft.partnercatalyst.fortis.spark.transforms.Tag
import org.apache.commons.collections4.trie.PatriciaTrie

import scala.collection.mutable.ListBuffer

@SerialVersionUID(100L)
class KeywordExtractor(keyWords: Seq[String]) extends Serializable {
  private val wordTokenizer = """\b""".r
  private val keywordTrie = new PatriciaTrie[Unit]()
  private val originalCase = Map[String, String](keyWords.map(s => (s.toLowerCase, s)): _*)

  originalCase.keys.foreach(keywordTrie.put(_, ()))

  def extractKeywords(text: String): List[Tag] = {

    def findMatches(segment: Seq[String]): Iterable[String] = {
      val sb = new StringBuilder()
      val result = ListBuffer[String]()

      val it = segment.iterator
      var prefix = ""
      while (it.hasNext && !keywordTrie.prefixMap(prefix).isEmpty) {
        prefix = sb.append(it.next()).mkString

        if (keywordTrie.containsKey(prefix)) {
          result.append(originalCase(prefix))
        }
      }

      result
    }

    val tokens = wordTokenizer.split(text.toLowerCase).toSeq
    tokens.tails.flatMap(findMatches(_).map(Tag(_, 1))).toList
  }
}
--- 
+++ 
@@ -6,12 +6,9 @@
 import scala.collection.mutable.ListBuffer
 
 @SerialVersionUID(100L)
-class KeywordExtractor(keyWords: Seq[String]) extends Serializable {
-  private val wordTokenizer = """\b""".r
-  private val keywordTrie = new PatriciaTrie[Unit]()
-  private val originalCase = Map[String, String](keyWords.map(s => (s.toLowerCase, s)): _*)
-
-  originalCase.keys.foreach(keywordTrie.put(_, ()))
+class KeywordExtractor(keywords: Seq[String]) extends Serializable {
+  @transient private lazy val wordTokenizer = """\b""".r
+  @transient private lazy val keywordTrie = initializeTrie(keywords)
 
   def extractKeywords(text: String): List[Tag] = {
 
@@ -24,9 +21,7 @@
       while (it.hasNext && !keywordTrie.prefixMap(prefix).isEmpty) {
         prefix = sb.append(it.next()).mkString
 
-        if (keywordTrie.containsKey(prefix)) {
-          result.append(originalCase(prefix))
-        }
+        Option(keywordTrie.get(prefix)).foreach(result.append(_))
       }
 
       result
@@ -35,4 +30,11 @@
     val tokens = wordTokenizer.split(text.toLowerCase).toSeq
     tokens.tails.flatMap(findMatches(_).map(Tag(_, 1))).toList
   }
+
+  private def initializeTrie(keywords: Seq[String]): PatriciaTrie[String] = {
+    val trie = new PatriciaTrie[String]()
+    keywords.foreach(k => trie.put(k.toLowerCase, k))
+
+    trie
+  }
 }
Use lazy and transient initialization in keyword extractor.

Specifically, only the list of search keywords is non-transient. In this way, the initialization of computable data structures used in the implementation is deferred to the worker nodes.
