Develop DeduplicateDocuments filter

lib/src/main/resources/pipeline/all_duplicate_paragraphs.conf

@@ -1,3 +1,4 @@
 filters: [
+  {"class": "DuplicateDocuments"},
   {"class": "DuplicateParagraphs", "limit": 2}
 ]

lib/src/main/scala/com/worksap/nlp/uzushio/lib/cleaning/Pipeline.scala

@@ -88,13 +88,13 @@ class PerParagraphFilter(val filter: ParagraphFilter) extends DocFilter {
     .copy(paragraphs = doc.paragraphs.map(filter.checkParagraph))
 }
 
-final class Pipeline(filers: Array[DocFilter]) extends Serializable {
+final class Pipeline(filters: Array[DocFilter]) extends Serializable {
   def applyFilters(doc: Document): Document = {
     var i = 0
-    val len = filers.length
+    val len = filters.length
     var state = doc
-    while (i < len && state.remove != null) {
-      val f = filers(i)
+    while (i < len && state.remove == null) {
+      val f = filters(i)
       state = f.checkDocument(state)
       i += 1
     }

lib/src/main/scala/com/worksap/nlp/uzushio/lib/filters/DeduplicateDocuments.scala

@@ -0,0 +1,69 @@
+package com.worksap.nlp.uzushio.lib.filters
+
+import com.worksap.nlp.uzushio.lib.stats.NgramHashExtractor
+import com.worksap.nlp.uzushio.lib.cleaning.Document
+import com.worksap.nlp.uzushio.lib.filters.base.DocFilter
+import com.worksap.nlp.uzushio.lib.utils.MathUtil
+import scala.math._
+import scala.util.Random
+
+trait RandomGeneratorFromStringBase {
+  def generateRandom(docId: String): Double
+}
+
+// An object in arguments of DocFilter on Spark needs to mixin Serializable.
+object RandomGeneratorFromString extends RandomGeneratorFromStringBase with Serializable {
+  def generateRandom(docId: String): Double = {
+    val seed = NgramHashExtractor.hashString(docId)
+    MathUtil.asRandomDouble(seed)
+  }
+}
+
+class GaussianRandomGeneratorFromString(
+    val mu: Double = 0.3,
+    val sd: Double = 0.1
+) extends RandomGeneratorFromStringBase
+    with Serializable {
+  def generateRandom(docId: String): Double = {
+    val seed = NgramHashExtractor.hashString(docId)
+    Random.setSeed(seed)
+    Random.nextGaussian() * mu + sd
+  }
+}
+
+class DeduplicateDocuments(
+    val baseNumFreq: Int = 10,
+    val randomGenerator: RandomGeneratorFromStringBase = new GaussianRandomGeneratorFromString
+) extends DocFilter {
+
+  def computeNearDuplicateTextRatio(doc: Document): Float = {
+    val iter = doc.aliveParagraphs
+
+    var totalLengthWeightedNearFreq = 0.0
+    var totalLength = 0.0
+
+    while (iter.hasNext) {
+      val paragraph = iter.next()
+      val text = paragraph.text
+      val textLength = text.length()
+      val nearFreq = if (paragraph.nearFreq < baseNumFreq) paragraph.nearFreq else baseNumFreq
+      val weight = log(nearFreq) / log(baseNumFreq)
+
+      totalLength += textLength
+      totalLengthWeightedNearFreq += (textLength * weight)
+    }
+
+    MathUtil.ratio(totalLengthWeightedNearFreq.toFloat, totalLength.toFloat)
+  }
+
+  def shouldRemoveDocument(doc: Document) = {
+    val nearDuplicateTextRatio = computeNearDuplicateTextRatio(doc)
+    val thresholdProb = randomGenerator.generateRandom(doc.render())
+
+    nearDuplicateTextRatio >= thresholdProb
+  }
+
+  override def checkDocument(doc: Document): Document = {
+    doc.removeWhen(shouldRemoveDocument(doc), this)
+  }
+}

lib/src/main/scala/com/worksap/nlp/uzushio/lib/runners/DedupFilterStatistics.scala

@@ -2,6 +2,7 @@ package com.worksap.nlp.uzushio.lib.runners
 
 import com.worksap.nlp.uzushio.lib.cleaning.Document
 import com.worksap.nlp.uzushio.lib.utils.Resources.AutoClosableResource
+import com.worksap.nlp.uzushio.lib.filters.DeduplicateDocuments
 import org.apache.spark.sql.expressions.UserDefinedFunction
 import org.apache.spark.sql.functions.{collect_list, octet_length, udf}
 import org.apache.spark.sql.{SaveMode, SparkSession}
@@ -42,7 +43,7 @@ object DedupFilterStatistics {
 
     val withValues = assembledDocs.select(
       metric($"docId", $"text", $"pos", $"exactFreq", $"nearFreq") as "res"
-    ).select(// make columns in the same order as FilterStatistics
+    ).select( // make columns in the same order as FilterStatistics
       $"res._2" as "value",
       $"res._1" as "text"
     )
@@ -52,6 +53,11 @@ object DedupFilterStatistics {
       .csv(args.output())
   }
 
+  def computeDuplicationScore(doc: Document, baseNumFreq: Int = 10) = {
+    val filter = new DeduplicateDocuments(baseNumFreq)
+    filter.computeNearDuplicateTextRatio(doc)
+  }
+
   def ratioUdfConstructor[T: TypeTag](sample: Double)(extractor: Document => Float): UserDefinedFunction = {
     udf {
       (
@@ -86,6 +92,8 @@ object DedupFilterStatistics {
         udfMaker(doc => doc.aliveParagraphs.map(_.nearFreq).foldRight(0)(_.max(_)))
       case "min-near-freq" =>
         udfMaker(doc => doc.aliveParagraphs.map(_.nearFreq).foldRight(0)(_.min(_)))
+      case "duplication-score" =>
+        udfMaker(doc => computeDuplicationScore(doc))
       case _ => throw new IllegalArgumentException(s"unknown metric $ftype")
     }
 

lib/src/main/scala/com/worksap/nlp/uzushio/lib/runners/DeduplicateParagraphs.scala

@@ -699,7 +699,10 @@ object DeduplicateParagraphs {
   ): String = {
     val doc = Document(parts)
     val filtered = args.pipeline.applyFilters(doc)
-    filtered.copy(paragraphs = doc.paragraphs.filter(_.remove != null)).render()
+    if (filtered.remove != null) {
+        return filtered.copy(IndexedSeq()).render()
+    }
+    filtered.copy(paragraphs = filtered.paragraphs.filter(_.remove == null)).render()
   }
 
   // noinspection TypeAnnotation,ScalaWeakerAccess

lib/src/main/scala/com/worksap/nlp/uzushio/lib/utils/MathUtil.java

@@ -95,6 +95,13 @@ public static int matchingBits(long x, long y) {
         return Long.bitCount(~(x ^ y));
     }
 
+    public static float ratio(float sum, float total) {
+        if (sum == 0 || total == 0) {
+            return 0.0f;
+        }
+        return sum / total;
+    }
+
     public static float ratio(int count, int total) {
         if (count == 0 || total == 0) {
             return 0.0f;

lib/src/test/scala/com/worksap/nlp/uzushio/lib/filters/DeduplicateDocumentsSpec.scala

@@ -0,0 +1,64 @@
+package com.worksap.nlp.uzushio.lib.filters
+
+import com.worksap.nlp.uzushio.lib.cleaning.Document
+import javax.swing.tree.FixedHeightLayoutCache
+import org.scalatest.freespec.AnyFreeSpec
+
+
+class FixedProbRandomGenerator(
+  val returnProb: Double = 0.5
+) extends RandomGeneratorFromStringBase {
+  def generateRandom(docId: String): Double = returnProb
+}
+
+
+class DeduplicateDocumentsSpec extends AnyFreeSpec {
+  def generateFilter(returnProb: Double): DeduplicateDocuments = {
+    val randomGenerator = new FixedProbRandomGenerator(returnProb)
+    new DeduplicateDocuments(100, randomGenerator)
+  }
+
+  "DeduplicateDocumentsSpec" - {
+    val filter = generateFilter(0.5)
+
+    "computes correct ratio for non-deuplicated documents" in {
+      val paragraphs = testParagraphs(
+        Seq("test", "test", "test", "test"),
+        Seq(1, 1, 1, 1)
+      )
+      val doc = Document(paragraphs, "test")
+      assert(0.0f == filter.computeNearDuplicateTextRatio(doc))
+      assert(false == filter.shouldRemoveDocument(doc))
+    }
+
+    "computes correct ratio for non-deuplicated documents (boundary)" in {
+      val paragraphs = testParagraphs(
+        Seq("test", "test", "test", "test"),
+        Seq(1, 1, 99, 100)
+      )
+      val doc = Document(paragraphs, "test")
+      assert(0.5f > filter.computeNearDuplicateTextRatio(doc))
+      assert(false == filter.shouldRemoveDocument(doc))
+    }
+
+    "computes correct ratio for deuplicated documents" in {
+      val paragraphs = testParagraphs(
+        Seq("test", "test", "test", "test"),
+        Seq(100, 100, 100, 100)
+      )
+      val doc = Document(paragraphs, "test")
+      assert(1.0f == filter.computeNearDuplicateTextRatio(doc))
+      assert(true == filter.shouldRemoveDocument(doc))
+    }
+
+    "computes correct ratio for deuplicated documents (boundary)" in {
+      val paragraphs = testParagraphs(
+        Seq("test", "test", "test", "test"),
+        Seq(1, 1, 100, 100)
+      )
+      val doc = Document(paragraphs, "test")
+      assert(0.5f == filter.computeNearDuplicateTextRatio(doc))
+      assert(true == filter.shouldRemoveDocument(doc))
+    }
+  }
+}

lib/src/test/scala/com/worksap/nlp/uzushio/lib/filters/package.scala

@@ -31,4 +31,11 @@ package object filters {
       }.toIndexedSeq
     )
   }
+
+  def testParagraphs(texts: Seq[String], nearFreqs: Seq[Int]): IndexedSeq[Paragraph] = {
+      (texts, nearFreqs)
+        .zipped
+        .map ((text, freq) => Paragraph("", text, 0, 1, freq))
+        .toIndexedSeq
+  }
 }