SDS-2.x, Scalable Data Engineering Science

spark.version

%sh 
curl -o /dbfs/FileStore/tables/pos_ud-gsd_fr_2.0.0_2.4_1553029753307.zip https://s3.amazonaws.com/auxdata.johnsnowlabs.com/public/models/pos_ud-gsd_fr_2.0.0_2.4_1553029753307.zip

%sh
ls -l "/dbfs/FileStore/tables/pos_ud-gsd_fr_2.0.0_2.4_1553029753307.zip"

%sh
unzip -o /dbfs/FileStore/tables/pos_ud-gsd_fr_2.0.0_2.4_1553029753307.zip -d /dbfs/FileStore/tables/

%sh
ls -l /dbfs/FileStore/tables/pos_ud-gsd_fr_2.0.0_2.4_1553029753307/

import org.apache.spark.mllib.classification.LogisticRegressionWithLBFGS
import org.apache.spark.mllib.evaluation.MulticlassMetrics
import org.apache.spark.mllib.evaluation.MultilabelMetrics
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.util.MLUtils
import org.apache.spark.ml._
import org.apache.spark.ml.feature.StringIndexer
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.types.MetadataBuilder
import org.apache.spark.sql.{Column, DataFrame, SparkSession}
import org.apache.spark.sql.functions.{col, concat_ws, lit, split, udf}
import org.apache.spark.ml.feature.NGram
import org.apache.spark.ml.Pipeline

import com.johnsnowlabs.nlp.{Annotation, AnnotatorType}
import com.johnsnowlabs.nlp.{DocumentAssembler, Finisher}
import com.johnsnowlabs.nlp.annotators.{Normalizer, Stemmer, Tokenizer}
import com.johnsnowlabs.nlp.annotator._
import com.johnsnowlabs.nlp.base._
import com.johnsnowlabs.util.Benchmark

import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.functions._

import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
import org.apache.hadoop.io.{LongWritable, Text}
import org.apache.spark.rdd.RDD
import scala.collection.mutable.ArrayBuffer

import scala.util.control.Breaks._

import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
import org.apache.hadoop.io.{LongWritable, Text}
import org.apache.spark.rdd.RDD

 def extractTokens = udf { docs: Seq[String] =>
    var posTagsArray = ArrayBuffer[String]()
    var previousSentenceNumber = Array[String]()
    for ((e,i) <- docs.zipWithIndex){
      val splitedArray = e.split("\t")      
      val currentSentenceNumber = splitedArray(0).split("-")

      if(currentSentenceNumber.length > 1){
        previousSentenceNumber = currentSentenceNumber
        val nextSentence = docs(i+1).split("\t")
        posTagsArray += splitedArray(1)
      }else if(previousSentenceNumber.contains(currentSentenceNumber(0))){

      }else{
        posTagsArray += splitedArray(1)
      }
    }
    posTagsArray
  }

  def extractTags = udf { docs: Seq[String] =>
    var posTagsArray = ArrayBuffer[String]()
    var previousSentenceNumber = Array[String]()
    for ((e,i) <- docs.zipWithIndex){
      val splitedArray = e.split("\t")

      val currentSentenceNumber = splitedArray(0).split("-")
      if(currentSentenceNumber.length > 1){
        previousSentenceNumber = currentSentenceNumber
        val nextSentence = docs(i+1).split("\t")
        posTagsArray += nextSentence(3)

      }else if(previousSentenceNumber.contains(currentSentenceNumber(0))){

      }else{
        posTagsArray += splitedArray(3)
      }
    }
    posTagsArray
  }
// these are for some checks afterwards
 def extractMissingTokens= udf { (testTokens: Seq[String], predictTokens: Seq[String]) =>
    var missingTokensArray = ArrayBuffer[String]()

    for (e <- testTokens) {
      if (!predictTokens.contains(e)) {
        missingTokensArray += e
      }
    }
    missingTokensArray
  }
  def calLengthOfArray = udf { docs: Seq[String] =>
    docs.length
  }

def customeTokenizer: UserDefinedFunction = udf { (tokens: Seq[String], text: String, sentenceIndex: String) =>
  lazy val strTokens = tokens.mkString("#")

  val tokenTagAnnotation: ArrayBuffer[Annotation] = ArrayBuffer()
  def annotatorType: String = AnnotatorType.TOKEN
  var lastIndex = 0

  for ((e, i) <- tokens.zipWithIndex) {

    val beginOfToken = text.indexOfSlice(e, lastIndex)
    val endOfToken = (beginOfToken + e.length) - 1

    val fullTokenAnnotatorStruct = new Annotation(
      annotatorType = annotatorType,
      begin=beginOfToken,
      end=endOfToken,
      result=e,
      metadata=Map("sentence" -> sentenceIndex)
    )
    tokenTagAnnotation += fullTokenAnnotatorStruct
    lastIndex = text.indexOfSlice(e, lastIndex)
  }
  tokenTagAnnotation
}
def wrapColumnMetadata(col: Column, annotatorType: String, outPutColName: String): Column = {
    val metadataBuilder: MetadataBuilder = new MetadataBuilder()
    metadataBuilder.putString("annotatorType", annotatorType)
    col.as(outPutColName, metadataBuilder.build)
}

%sh
wget -N https://github.com/UniversalDependencies/UD_French-GSD/raw/master/fr_gsd-ud-test.conllu -P /dbfs/FileStore/tables/

%sh
ls -l /dbfs/FileStore/tables/fr_gsd-ud-test.conllu

import org.apache.hadoop.io.LongWritable
import org.apache.hadoop.io.Text
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat

val pathCoNNLFile = "/FileStore/tables/fr_gsd-ud-test.conllu"

// We need to change the Hadoop's default delimiter from 1 newline to 2 newlines
// This way we can catch CoNLL-U 2x newlines that seperates sentences
// This will show deprecation warning, don't mind it for the time being :)
// NOTE: we should make these transient to avoid task not serializable
@transient val conf = new org.apache.hadoop.mapreduce.Job().getConfiguration
conf.set("textinputformat.record.delimiter", "\n\n")

@transient val usgRDD = sc.newAPIHadoopFile(pathCoNNLFile, classOf[TextInputFormat], classOf[LongWritable], classOf[Text], conf).map{ case (_, v) => v.toString }
  
@transient val conllSentencesDF = usgRDD.map(s => s.split("\n").filter(x => !x.startsWith("#"))).filter(x => x.length > 0).toDF("sentence")

conf.set("textinputformat.record.delimiter", "\n")

// testTokensTagsDF will be test dataset's sentences, tokens, and tags.
val testTokensTagsDF = conllSentencesDF
.withColumn("id", monotonically_increasing_id)
.withColumn("testTokens", extractTokens($"sentence"))
.withColumn("testTags", extractTags($"sentence"))
.withColumn("text", concat_ws(" ", $"testTokens"))
.drop("sentence")

testTokensTagsDF.show()

// Annotate the documents
// We don't need SentenceDetector nor Tokenizer since in this example we'll use golden sentences/tokens

val documentAssembler = new DocumentAssembler()
    .setInputCol("text")
    .setOutputCol("document")
    .transform(testTokensTagsDF)

// create Tokenizer column based on golden tokens
// we only select document, sentence and tokens to feed into our POS Model for prediction
val goldenTokenizer = documentAssembler
.withColumn("documentText", $"document.result"(0))
.withColumn("sentenceIndex", lit("0"))
.withColumn("token", customeTokenizer($"testTokens", $"documentText", $"sentenceIndex"))
.withColumn("token", wrapColumnMetadata($"token", AnnotatorType.TOKEN, "token"))
.select("id", "document", "token")

// Let's check make sure we got Tokenizer column right
val datasetSchemaFields = goldenTokenizer.schema.fields
        .find(f => f.metadata.contains("annotatorType") && f.metadata.getString("annotatorType") == AnnotatorType.TOKEN)

val tokenColumn = datasetSchemaFields.map(_.name).get

// Let's load our pretrained POS Model
// As you can see I am only using the document and token columns to output prediction column called pos
val perceptronModel = PerceptronModel.load("/FileStore/tables/pos_ud-gsd_fr_2.0.0_2.4_1553029753307/")
    .setInputCols(Array("document", "token"))
    .setOutputCol("pos")
    .transform(goldenTokenizer)
    .select(
        $"id",
        $"document",
        $"token",
        $"token.result".alias("predictedTokens"),
        $"pos.result".alias("predictedTags")
    )

perceptronModel.show

val joinedDF = perceptronModel
  .join(testTokensTagsDF, Seq("id"))
  .withColumn("predictedTokensLength", calLengthOfArray($"predictedTokens"))
  .withColumn("predictedTagsLength", calLengthOfArray($"predictedTags"))
  .withColumn("testTokensLength", calLengthOfArray($"testTokens"))
  .withColumn("testTagsLength", calLengthOfArray($"testTags"))
  .withColumn("tokensDiffFromTest", $"testTokensLength" - $"predictedTokensLength")
  .withColumn("tagsDiffFromTest", $"testTagsLength" - $"predictedTagsLength")
  .withColumn("missingTokens", extractMissingTokens($"testTokens", $"predictedTokens"))
  .withColumn("missingTags", extractMissingTokens($"testTags", $"predictedTags"))
  .withColumn("equalTags", col("predictedTagsLength") === col("testTagsLength"))

joinedDF.agg(
    sum("predictedTokensLength").as("p_token_lengh"),
    sum("predictedTagsLength").as("p_tags_lengh"),
    sum("testTokensLength").as("t_token_lengh"),
    sum("predictedTagsLength").as("t_tags_lengh"),
    sum("tokensDiffFromTest").as("tokensDiffFromTest"),
    sum("tagsDiffFromTest").as("tagsDiffFromTest")
    ).show

spark.catalog.clearCache
joinedDF.count

case class NewColumns(label: String, prediction: String, labelIndex: Double, predictionIndex: Double)

val finalPredictionDF = joinedDF.select("testTokens", "testTags", "predictedTokens", "predictedTags").flatMap(row => {
    val labelsMap: List[(String, Double)] = List(
        ("ADJ",1.0), ("ADP",2.0), ("ADV",3.0),
        ("AUX",4.0), ("CCONJ",5.0), ("DET",6.0),
        ("INTJ",7.0), ("NOUN",8.0), ("NUM",9.0),
        ("PART",10.0), ("PRON",11.0), ("PROPN",12.0),
        ("PUNCT",13.0), ("SCONJ",14.0), ("SYM",15.0),
        ("VERB",16.0), ("X",17.0)
    )
    val finalRow: ArrayBuffer[Seq[NewColumns]] = ArrayBuffer()
    val newRow: ArrayBuffer[NewColumns] = ArrayBuffer()
    
    val testTagsWithTokens = row.get(0).asInstanceOf[Seq[String]].zip(row.getSeq(1).asInstanceOf[Seq[String]]).map{case (k,v) => (k,v)}
    var predictTagsWithTokens = row.getSeq(2).asInstanceOf[Seq[String]].zip(row.getSeq(3).asInstanceOf[Seq[String]]).map{case (k,v) => (k,v)}
    
    val testTags = row.getSeq(1).asInstanceOf[Seq[String]]
    val predictTags = row.getSeq(3).asInstanceOf[Seq[String]]
       
    for ((t,i) <- testTags.zipWithIndex) {

        val labelIndex = labelsMap.find(_._1.matches(t)).map(_._2).getOrElse(0.0).asInstanceOf[Double]
        val predictionIndex = labelsMap.find(_._1.matches(predictTags(i))).map(_._2).getOrElse(0.0).asInstanceOf[Double]
        
        newRow += NewColumns(t, predictTags(i), labelIndex, predictionIndex)
    }
    finalRow.append(newRow)
    finalRow
})
.select(explode($"value").as("valueArray"))
.withColumn("label", $"valueArray.label")
.withColumn("prediction", $"valueArray.prediction")
.withColumn("labelIndex", $"valueArray.labelIndex")
.withColumn("predictionIndex", $"valueArray.predictionIndex")
.drop("value", "valueArray")

finalPredictionDF.count
finalPredictionDF.show(2, false)
finalPredictionDF.printSchema  

//convert our DataFrame to 2 columns RDD with double values
val predictionLabelsRDD = finalPredictionDF.select("predictionIndex", "labelIndex").map(r => (r.getDouble(0), r.getDouble(1)))
val metrics = new MulticlassMetrics(predictionLabelsRDD.rdd)

// we use this to convert the index back to its label
val labelsMap: List[(String, Double)] = List(
        ("ADJ",1.0), ("ADP",2.0), ("ADV",3.0),
        ("AUX",4.0), ("CCONJ",5.0), ("DET",6.0),
        ("INTJ",7.0), ("NOUN",8.0), ("NUM",9.0),
        ("PART",10.0), ("PRON",11.0), ("PROPN",12.0),
        ("PUNCT",13.0), ("SCONJ",14.0), ("SYM",15.0),
        ("VERB",16.0), ("X",17.0)
)

// Overall Statistics
val accuracy = metrics.accuracy
println("Summary Statistics")
println(s"Accuracy = $accuracy")

val labels = metrics.labels

// Precision by label
labels.foreach { l =>
  val labelStr = labelsMap.find(_._2 == l).map(_._1).getOrElse("Not defined").asInstanceOf[String]
  println(s"Precision($labelStr) = " + metrics.precision(l))
}