Sparkling integration updates.

* fix discardDate issue
* update tests for #494
* add test for #493
* add test for #532
* move issue specific tests to their own directory
* add copyright statement to SparklingArchiveRecord
* move webarchive-commons back to 1.1.9
* resolves #532
* resolves #494
* resolves #493
* resolves #492
* resolves #317
* resolves #260
* resolves #182
* resolves #76
* resolves #74
* resolves #73
* resolves #23
* resolves #18

Author: ruebot

pom.xml

@@ -107,14 +107,12 @@
                   <resource>META-INF/services/org.apache.lucene.codecs.Codec</resource>
                 </transformer>
               </transformers>
-
               <relocations>
                 <relocation>
                   <pattern>com.google.common.</pattern>
                   <shadedPattern>com.google.common.shaded.</shadedPattern>
                 </relocation>
               </relocations>
-
               <!-- This fixes the issue "Invalid signature file digest for Manifest main attributes"
                    cf. http://zhentao-li.blogspot.com/2012/06/maven-shade-plugin-invalid-signature.html -->
               <filters>
@@ -483,7 +481,7 @@
     <dependency>
       <groupId>org.netpreserve.commons</groupId>
       <artifactId>webarchive-commons</artifactId>
-      <version>1.1.8</version>
+      <version>1.1.9</version>
       <exclusions>
         <exclusion>
           <groupId>org.apache.hadoop</groupId>

src/main/scala/io/archivesunleashed/SparklingArchiveRecord.scala

@@ -1,3 +1,19 @@
+/*
+ * Copyright © 2017 The Archives Unleashed Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package io.archivesunleashed
 
 import java.io.InputStream

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

@@ -814,8 +814,11 @@ package object archivesunleashed {
       *
       * @param date a list of dates
       */
-    def discardDate(date: String): RDD[ArchiveRecord] = {
-      rdd.filter(r => r.getCrawlDate != date)
+    def discardDate(
+      dates: List[String],
+      component: DateComponent = DateComponent.YYYYMMDD
+    ): RDD[ArchiveRecord] = {
+      rdd.filter(r => !dates.contains(ExtractDate(r.getCrawlDate, component)))
     }
 
     /** Filters detected URLs.

src/test/resources/warc/issue-493.warc

@@ -27,7 +27,6 @@ import org.scalatest.{BeforeAndAfter, FunSuite}
 @RunWith(classOf[JUnitRunner])
 class RecordRDDTest extends FunSuite with BeforeAndAfter {
   private val arcPath = Resources.getResource("arc/example.arc.gz").getPath
-  private val badPath = Resources.getResource("arc/badexample.arc.gz").getPath
   private val master = "local[4]"
   private val appName = "example-spark"
   private var sc: SparkContext = _
@@ -43,24 +42,6 @@ class RecordRDDTest extends FunSuite with BeforeAndAfter {
     sc = new SparkContext(conf)
   }
 
-  test("Expect no valid pages RDD") {
-    val expectedLength = 0
-    val base = RecordLoader
-      .loadArchives(badPath, sc)
-      .keepValidPages()
-      .take(2)
-    assert(base.length == expectedLength)
-  }
-
-  test("Expect no images RDD") {
-    val expectedLength = 0
-    val base = RecordLoader
-      .loadArchives(badPath, sc)
-      .keepValidPages()
-      .take(2)
-    assert(base.length == expectedLength)
-  }
-
   test("Keep date RDD") {
     val testDate = "2008"
     val base = RecordLoader.loadArchives(arcPath, sc)
@@ -220,10 +201,17 @@ class RecordRDDTest extends FunSuite with BeforeAndAfter {
 
   test("Discard date RDD") {
     val base = RecordLoader.loadArchives(arcPath, sc)
-    val date = "20080430"
-    val r = base.filter(x => !(x.getCrawlDate.contains(date))).collect()
-    val r2 = base.discardDate(date).take(3)
-    assert(r.deep == Array().deep)
+    val date = "2007"
+    val dateComponent = DateComponent.YYYY
+    val r = base
+      .filter(x => ExtractDate(x.getCrawlDate, dateComponent) != date)
+      .map(mp => mp.getUrl)
+      .take(3)
+    val r2 = base
+      .discardDate(List(date), dateComponent)
+      .map(mp => mp.getUrl)
+      .take(3)
+    assert(r2.sameElements(r))
   }
 
   test("Discard URLs RDD") {

src/test/scala/io/archivesunleashed/RecordRDDTest.scala

@@ -0,0 +1,66 @@
+/*
+ * Copyright © 2017 The Archives Unleashed Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.archivesunleashed.app
+
+import com.google.common.io.Resources
+import io.archivesunleashed.RecordLoader
+import io.archivesunleashed.udfs.{removeHTML, removeHTTPHeader}
+import org.apache.spark.sql.SparkSession
+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 Issue493Test extends FunSuite with BeforeAndAfter {
+  private val arcPath = Resources.getResource("warc/issue-493.warc").getPath
+  private val master = "local[4]"
+  private val appName = "example-spark"
+  private var sc: SparkContext = _
+
+  before {
+    val conf = new SparkConf()
+      .setMaster(master)
+      .setAppName(appName)
+    sc = new SparkContext(conf)
+  }
+
+  test("Test for issue 493 - compressed payload warcs") {
+    val df = RecordLoader.loadArchives(arcPath, sc).webpages()
+
+    // We need this in order to use the $-notation
+    val spark = SparkSession.builder().master("local").getOrCreate()
+    // scalastyle:off
+    import spark.implicits._
+    // scalastyle:on
+
+    val dfResults = df
+      .select(removeHTML(removeHTTPHeader($"content")))
+      .head(2)
+    val RESULTSLENGTH = 2
+
+    assert(dfResults.length == RESULTSLENGTH)
+    assert(dfResults(0).get(0) == "makkaronisch fuer niedlich")
+    assert(dfResults(1).get(0) == "makkaronisch fuer niedlich die melodie")
+  }
+
+  after {
+    if (sc != null) {
+      sc.stop()
+    }
+  }
+}