Add more DataFrame Implementation Serializable APIs (#401).

- Partially addresses  #223 
- Add discardContentDF
- Add discardUrlPatternsDF
- Add discardLanguagesDF
- Add keepImagesDF
- Add keepContentDF
- Add keepUrlPatternsDF
- Add keepLanguagesDF
- Update tests

src/main/scala/io/archivesunleashed/package.scala

@@ -23,7 +23,7 @@ import io.archivesunleashed.data.ArchiveRecordWritable.ArchiveFormat
 import io.archivesunleashed.data.{ArchiveRecordInputFormat, ArchiveRecordWritable}
 
 import ArchiveRecordWritable.ArchiveFormat
-import io.archivesunleashed.df.{DetectMimeTypeTikaDF, ExtractDateDF, ExtractDomainDF}
+import io.archivesunleashed.df.{DetectLanguageDF, DetectMimeTypeTikaDF, ExtractDateDF, ExtractDomainDF, RemoveHTMLDF}
 
 import io.archivesunleashed.matchbox.{DetectLanguageRDD, DetectMimeTypeTika, ExtractDateRDD,
                                       ExtractDomainRDD, ExtractImageDetails, ExtractImageLinksRDD,
@@ -158,6 +158,51 @@ package object archivesunleashed {
       df.filter(filteredHttpStatus($"HttpStatus"))
     }
 
+    /** Filters detected content (regex).
+      *
+      * @param contentREs a list of regular expressions
+      */
+    def discardContentDF(contentREs: Set[Regex]): DataFrame = {
+      val filteredContent = udf((c: String) => {
+                          !contentREs.map(re =>
+                          (re findFirstIn c) match {
+                            case Some(v) => true
+                            case None => false
+                          }).exists(identity)
+                        })
+      df.filter(filteredContent($"content"))
+    }
+
+    /** Filters detected URL patterns (regex).
+     *
+     *  @param urlREs a list of Regular expressions
+     */
+    def discardUrlPatternsDF(urlREs: Set[Regex]): DataFrame = {
+      val filteredUrlPatterns = udf((urlPattern: String) => {
+                              !urlREs.map(re =>
+                                urlPattern match {
+                                  case re() => true
+                                  case _ => false
+                              }).exists(identity)
+                            })
+      df.filter(filteredUrlPatterns($"url"))
+    }
+
+    /** Filters detected language.
+      *
+      * @param lang a set of ISO 639-2 codes
+      */
+    def discardLanguagesDF(lang: Set[String]): DataFrame = {
+      val filteredLanguage = udf((language: String) => !lang.contains(language))
+      df.filter(filteredLanguage(DetectLanguageDF(RemoveHTMLDF($"content"))))
+    }
+
+    /** Removes all data except images. */
+    def keepImagesDF(): DataFrame = {
+      val takeImages = udf((date: String, mimeType: String) => date != null && mimeType.startsWith("image/"))
+      df.filter(takeImages($"crawl_date", DetectMimeTypeTikaDF($"bytes")))
+    }
+
     /** Removes all data that does not have selected HTTP status codes.
      *
      *  @param statusCodes a list of HTTP status codes
@@ -212,6 +257,45 @@ package object archivesunleashed {
       val takeMimeType = udf((mimeType: String) => mimeTypes.contains(mimeType))
       df.filter(takeMimeType($"mime_type_web_server"))
     }
+
+    /** Removes all content that does not pass Regular Expression test.
+      *
+      * @param contentREs a list of regular expressions to keep
+      */
+    def keepContentDF(contentREs: Set[Regex]): DataFrame = {
+      val takeContent = udf((c: String) => {
+                          contentREs.map(re =>
+                            (re findFirstIn c) match {
+                              case Some(v) => true
+                              case None => false
+                          }).exists(identity)
+                        })
+      df.filter(takeContent($"content"))
+    }
+
+    /** Removes all data but selected URL patterns.
+      *
+      * @param urlREs a list of regular expressions
+      */
+    def keepUrlPatternsDF(urlREs: Set[Regex]): DataFrame = {
+      val takeUrlPatterns = udf((urlPattern: String) => {
+                              urlREs.map(re =>
+                                urlPattern match {
+                                  case re() => true
+                                  case _ => false
+                              }).exists(identity)
+                            })
+      df.filter(takeUrlPatterns($"url"))
+    }
+
+    /** Removes all data not in selected language.
+      *
+      * @param lang a set of ISO 639-2 codes
+      */
+    def keepLanguagesDF(lang: Set[String]): DataFrame = {
+      val takeLanguage = udf((language: String) => lang.contains(language))
+      df.filter(takeLanguage((DetectLanguageDF(RemoveHTMLDF($"content")))))
+    }
   }
 
   /**
@@ -815,4 +899,4 @@ package object archivesunleashed {
       rdd.filter(r => !lang.contains(DetectLanguageRDD(RemoveHTMLRDD(r.getContentString))))
     }
   }
-}
+}

src/test/scala/io/archivesunleashed/RecordDFTest.scala

@@ -16,12 +16,14 @@
 
 package io.archivesunleashed
 
+import io.archivesunleashed.df.{DetectLanguageDF, RemoveHTMLDF}
 import com.google.common.io.Resources
 import org.apache.spark.{SparkConf, SparkContext}
 import org.junit.runner.RunWith
 import org.scalatest.junit.JUnitRunner
 import org.scalatest.{BeforeAndAfter, FunSuite}
 
+
 @RunWith(classOf[JUnitRunner])
 class RecordDFTest extends FunSuite with BeforeAndAfter {
   private val arcPath = Resources.getResource("arc/example.arc.gz").getPath
@@ -89,6 +91,34 @@ class RecordDFTest extends FunSuite with BeforeAndAfter {
     assert (base.toString == expected)
   }
 
+  test("Discard Content") {
+    val expected = "dns:www.archive.org"
+    val reg = Set("Content-Length: [0-9]{4}".r)
+    val base = RecordLoader.loadArchives(arcPath, sc).all()
+      .select("url", "content").discardContentDF(reg).take(2)(1)(0)
+
+    assert (base.toString == expected)
+  }
+
+  test("Discard UrlPatterns") {
+    val expected = "dns:www.archive.org"
+    val reg = Set(".*images.*".r)
+    val base = RecordLoader.loadArchives(arcPath, sc).all()
+      .select("url").discardUrlPatternsDF(reg).take(2)(1)(0)
+
+    assert (base.toString == expected)
+  }
+
+  test("Discard Languages") {
+    val expected = "dns:www.archive.org"
+    val reg = Set("th","de","ht")
+    val base = RecordLoader.loadArchives(arcPath, sc).all()
+      .select("url")
+      .discardLanguagesDF(reg).take(2)(1)(0)
+
+    assert (base.toString == expected)
+  }
+
   test("Keep HttpStatus") {
     val expected = "http://www.archive.org/robots.txt"
     val statusCodes = Set("200")
@@ -125,7 +155,7 @@ class RecordDFTest extends FunSuite with BeforeAndAfter {
     assert (base.toString == expected)
   }
 
-  test("Keep keepMimeTypesTika") {
+  test("Keep MimeTypesTika") {
     val expected = "image/jpeg"
     val domains = Set("image/jpeg")
     val base = RecordLoader.loadArchives(arcPath, sc).all()
@@ -134,7 +164,7 @@ class RecordDFTest extends FunSuite with BeforeAndAfter {
     assert (base.toString == expected)
   }
 
-  test("Keep keepMimeTypes") {
+  test("Keep MimeTypes") {
     val expected = "text/html"
     val domains = Set("text/html")
     val base = RecordLoader.loadArchives(arcPath, sc).all()
@@ -143,6 +173,45 @@ class RecordDFTest extends FunSuite with BeforeAndAfter {
     assert (base.toString == expected)
   }
 
+  test("Keep content") {
+    val expected = "http://www.archive.org/images/logoc.jpg"
+    val reg = Set("Content-Length: [0-9]{4}".r)
+    val base = RecordLoader.loadArchives(arcPath, sc).all()
+      .select("url", "content")
+      .keepContentDF(reg).take(1)(0)(0)
+
+    assert (base.toString == expected)
+  }
+
+  test("Keep UrlPatterns") {
+    val expected = "http://www.archive.org/images/go-button-gateway.gif"
+    val reg = Set("text/html")
+    val base = RecordLoader.loadArchives(arcPath, sc).all()
+      .select("url")
+      .keepUrlPatternsDF(Set(".*images.*".r)).take(2)(1)(0)
+
+    assert (base.toString == expected)
+  }
+
+  test("Keep Languages") {
+    val expected = "http://www.archive.org/images/logoc.jpg"
+    val reg = Set("th","de","ht")
+    val base = RecordLoader.loadArchives(arcPath, sc).all()
+      .select("url")
+      .keepLanguagesDF(reg).take(1)(0)(0)
+
+    assert (base.toString == expected)
+  }
+
+  test("Keep keepMimeTypes") {
+    val expected = "image/jpeg"
+    val base = RecordLoader.loadArchives(arcPath, sc).all()
+      .keepImagesDF()
+      .select("mime_type_tika").take(1)(0)(0)
+
+    assert (base.toString == expected)
+  }
+
   after {
     if (sc != null) {
       sc.stop()