We will demonstrate here how to bootstrap a simple Naive Bayes classification model using the home brewing related dataset from the Stack Exchange website:

$ wget https://archive.org/download/stackexchange/beer.stackexchange.com.7z
$ 7z e beer.stackexchange.com.7z

We create a few methods that pull both the body and labels from all XML documents, extract the clean text content out of the HTML encoded body (using the Goose scraper introduced in Chapter 6, Scraping Link-Based External Data) and finally convert our RDD of XML documents into a Spark DataFrame. The different methods are not reported here, but they can be found in our code repository. One needs to note that Goose scraper can be used offline by providing the HTML content (as a string) alongside a dummy URL.

We provide the reader with a convenient parse method that can be used for pre-processing any Post.xml data from the Stack Exchange website. This function is part of our StackBootstraping code available in our code repository:

import io.gzet.tagging.stackoverflow.StackBootstraping

val spark = SparkSession.builder()
  .appName("StackExchange")
  .getOrCreate()

val sc = spark.sparkContext
val rdd = sc.textFile("/path/to/posts.xml")
val brewing = StackBootstraping.parse(rdd)

brewing.show(5)

+--------------------+--------------------+
|                body|                tags|
+--------------------+--------------------+
|I was offered a b...|              [hops]|
|As far as we know...|           [history]|
|How is low/no alc...|           [brewing]|
|In general, what'...|[serving, tempera...|
|Currently I am st...| [pilsener, storage]|
+--------------------+--------------------+

With our beer content properly labeled, the remaining process is to bootstrap the algorithm itself. For that purpose, we use a simple Naive Bayes classification algorithm that determines the conditional probability of a label given an item's features. We first collect all distinct labels, assign a unique identifier (as Double), and broadcast our label dictionary to the Spark executors:

val labelMap = brewing
  .select("tags")
  .withColumn("tag", explode(brewing("tags")))
  .select("tag")
  .distinct()
  .rdd
  .map(_.getString(0)).zipWithIndex()
  .mapValues(_.toDouble + 1.0d)
labelMap.take(5).foreach(println)

/*
(imperal-stout,1.0)
(malt,2.0)
(lent,3.0)
(production,4.0)
(local,5.0)
*/

A LabeledPoint is composed of both a label (as Double) and features (as Vector). A common practice to build features out of text content is to build term frequency vectors, where each word across all documents corresponds to a specific dimension. With around hundreds of thousands of dimensions (the estimated number of words in English is 1,025,109), this highly dimensional space will be particularly inefficient for most machine learning algorithms. In fact, when Naive Bayes multiplies probabilities (lower than 1), there is a certain risk of reaching 0 due to machine precision issue (numerical underflow as described in Chapter 14, Scalable Algorithm). Data scientists overcome that constraint using the principle of dimensionality reduction, projecting a sparse vector into a denser space while preserving distance measures (the principle of dimensionality reduction will be covered in Chapter 10, Story De-duplication and Mutation). Although we can find many algorithms and techniques for that purpose, we will use the hashing utility provided by Spark.

With a vector size of n (default of 2²⁰), its transform method groups all words in n different buckets in respect to their hash values, and sums up the bucket frequencies to build denser vectors.

Prior to a dimensionality reduction, which can be an expensive operation, vector size can be greatly reduced by stemming and cleaning the text content. We use the Apache Lucene analyzer here:

<dependency>
   <groupId>org.apache.lucene</groupId>
   <artifactId>lucene-analyzers-common</artifactId>
   <version>4.10.1</version>
 </dependency>

We remove all punctuation and numbers and feed the plain text object to a Lucene analyzer, collecting each clean word as a CharTermAttribute:

def stem(rdd: RDD[(String, Array[String])]) = {
 
  val replacePunc = """\W""".r
  val replaceDigitOnly = """\s\d+\s""".r
 
  rdd mapPartitions { it =>

    val analyzer = new EnglishAnalyzer
    it map { case (body, tags) =>
      val content1 = replacePunc.replaceAllIn(body, " ")
      val content = replaceDigitOnly.replaceAllIn(content1, " ")
      val tReader = new StringReader(content)
      val tStream = analyzer.tokenStream("contents", tReader)
      val term = tStream.addAttribute(classOf[CharTermAttribute])
       tStream.reset()
      val terms = collection.mutable.MutableList[String]()
      while (tStream.incrementToken) {
        val clean = term.toString
        if (!clean.matches(".*\d.*") && clean.length > 3) {
           terms += clean
        }
      }
      tStream.close()
      (terms.toArray, tags)
     }

  }

With this approach, we transform the text [Mastering Spark for Data Science - V1] into [master spark data science], hence reducing the number of words (therefore dimensions) from our input vectors. Finally, we normalize our term frequency vector using the MLlib normalizer class:

val hashingTf = new HashingTF()
val normalizer = new Normalizer()

val labeledCorpus = stem(df map { row =>
  val body = row.getString(0)
  val tags = row.getAs[mutable.WrappedArray[String]](1)
  (body, tags)
})

val labeledPoints = labeledCorpus flatMap { case (corpus, tags) =>
  val vector = hashingTf.transform(corpus)
  val normVector = normalizer.transform(vector)
  tags map { tag =>
    val label = bLabelMap.value.getOrElse(tag, 0.0d)
    LabeledPoint(label, normVector)
  }
}

We train a Naive Bayes algorithm as follows and test our classifier using a test dataset that we did not include in the training data points. We finally display the first five predictions in the following example. The labels on the left-hand side are the original labels from our test content; on the right-hand side are the results of the Naive Bayes classification. An ipa has been predicted as hangover, validating with certainty the accuracy of our classification algorithm:

labeledPoints.cache()
val model: NaiveBayesModel = NaiveBayes.train(labeledPoints)
labeledPoints.unpersist(blocking = false)

model
  .predict(testPoints)
  .map { prediction =>
     bLabelMap.value.map(_.swap).get(prediction).get
   }
  .zip(testLabels)
  .toDF("predicted","original")
  .show(5)

+---------+-----------+
| original|  predicted|
+---------+-----------+
|  brewing|    brewing|
|      ipa|   hangover|
| hangover|   hangover|
| drinking|   drinking|
| pilsener|   pilsener|
+---------+-----------+

For convenience, we abstract all these methods and expose the following ones within a Classifier object that will be used later:

def train(rdd: RDD[(String, Array[String])]): ClassifierModel
def predict(rdd: RDD[String]): RDD[String]

We have demonstrated how to export labeled data from external sources, how to build a term frequency vector, and how to train a simple Naive Bayes classification model. The high level workflow used here is represented in the following figure and is common for most classification use cases:

Figure 1: Classification workflow

The next step is to start classifying the original unlabeled data (assuming our content is still brewery related). This closes the introduction of Naive Bayes classification and how a bootstrapped model could steal ground truth from external resources. Both these techniques will be used in our classification system in the following section.