Fixed bugs in mr snapshot, added changed factDiff to work on snapshot sequence files

Author: Russell Aronson

ivory-cli/src/main/scala/com/ambiata/ivory/cli/factDiff.scala

@@ -6,24 +6,23 @@ import com.ambiata.ivory.validate._
 
 object factDiff extends IvoryApp {
 
-  case class CliArguments(input1: String, input2: String, output: String, errors: String)
+  case class CliArguments(input1: String, input2: String, output: String)
 
   val parser = new scopt.OptionParser[CliArguments]("fact-diff") {
     head("""
-         |Compute diff between two files containing facts
+         |Compute diff between two sets of sequence files containing facts
          |""".stripMargin)
 
     help("help") text "shows this usage text"
-    opt[String]("input1") action { (x, c) => c.copy(input1 = x) } required() text s"Hdfs location to the first facts file."
-    opt[String]("input2") action { (x, c) => c.copy(input2 = x) } required() text s"Hdfs location to the first facts file."
+    opt[String]("input1") action { (x, c) => c.copy(input1 = x) } required() text s"Hdfs glob path to the first facts."
+    opt[String]("input2") action { (x, c) => c.copy(input2 = x) } required() text s"Hdfs glob path to the second facts."
     opt[String]('o', "output") action { (x, c) => c.copy(output = x) } required() text s"Hdfs location to store the difference."
-    opt[String]('e', "errors") action { (x, c) => c.copy(errors = x) } required() text s"Hdfs location to store any errors."
   }
 
-  val cmd = IvoryCmd[CliArguments](parser, CliArguments("", "", "", ""), ScoobiCmd { configuration => c =>
-      val res = FactDiff.scoobiJob(c.input1, c.input2, c.output, c.errors)
+  val cmd = IvoryCmd[CliArguments](parser, CliArguments("", "", ""), ScoobiCmd { configuration => c =>
+      val res = FactDiff.flatFacts(c.input1, c.input2, c.output)
       res.run(configuration).map {
-        case _ => List(s"Any differences can be found in '${c.output}', errors in '${c.errors}'")
+        case _ => List(s"Any differences can be found in '${c.output}'")
       }
     })
 }

ivory-extract/src/main/scala/com/ambiata/ivory/extract/SnapshotMr.scala

@@ -297,20 +297,32 @@ class SnapshotReducer extends Reducer[Text, BytesWritable, NullWritable, BytesWr
   val vout = new BytesWritable
 
   override def reduce(key: Text, iter: JIterable[BytesWritable], context: Reducer[Text, BytesWritable, NullWritable, BytesWritable]#Context): Unit = {
+
+    /****************** !!!!!! WARNING !!!!!! ******************
+     *
+     * This is some nasty nasty mutation that can coorrupt data
+     * without knowing, so double/triple check with others when
+     * changing.
+     *
+     ***********************************************************/
     val iterator = iter.iterator
     var latestContainer: PrioritizedFactBytes = null
-    var latestDate: Int = 0
+    var latestDate = 0l
+    var isTombstone = true
     while (iterator.hasNext) {
       val next = iterator.next
       deserializer.deserialize(container, next.getBytes)
       deserializer.deserialize(fact, container.getFactbytes)
-      val nextDate = fact.getYyyyMMdd
+      val nextDate = fact.datetime.long
       if(latestContainer == null || nextDate > latestDate || (nextDate == latestDate && container.getPriority < latestContainer.getPriority)) {
-        latestContainer = container
+        latestContainer = container.deepCopy
         latestDate = nextDate
+        isTombstone = fact.isTombstone
       }
     }
-    vout.set(latestContainer.getFactbytes, 0, latestContainer.getFactbytes.length)
-    context.write(NullWritable.get, vout)
+    if(!isTombstone) {
+      vout.set(latestContainer.getFactbytes, 0, latestContainer.getFactbytes.length)
+      context.write(NullWritable.get, vout)
+    }
   }
 }

ivory-validate/src/main/scala/com/ambiata/ivory/validate/FactDiff.scala

@@ -12,13 +12,36 @@ import FactFormats._
 
 object FactDiff {
 
-  def scoobiJob(input1: String, input2: String, outputPath: String, errorPath: String): ScoobiAction[Unit] = {
+  def partitionFacts(input1: String, input2: String, outputPath: String): ScoobiAction[Unit] = for {
+    res <- ScoobiAction.scoobiJob({ implicit sc: ScoobiConfiguration =>
+             val dlist1 = PartitionFactThriftStorageV1.PartitionedFactThriftLoader(List(input1)).loadScoobi.map({
+               case -\/(e) => sys.error(s"Can not parse fact - ${e}")
+               case \/-(f) => f
+             })
+             val dlist2 = PartitionFactThriftStorageV1.PartitionedFactThriftLoader(List(input2)).loadScoobi.map({
+               case -\/(e) => sys.error(s"Can not parse fact - ${e}")
+               case \/-(f) => f
+             })
+             (dlist1, dlist2)
+           })
+    (dlist1, dlist2) = res
+    _   <- scoobiJob(dlist1, dlist2, outputPath)
+  } yield ()
+
+  def flatFacts(input1: String, input2: String, outputPath: String): ScoobiAction[Unit] = for {
+    res <- ScoobiAction.scoobiJob({ implicit sc: ScoobiConfiguration =>
+             val dlist1 = valueFromSequenceFile[Fact](input1)
+             val dlist2 = valueFromSequenceFile[Fact](input2)
+             (dlist1, dlist2)
+           })
+    (dlist1, dlist2) = res
+    _   <- scoobiJob(dlist1, dlist2, outputPath)
+  } yield ()
+
+  def scoobiJob(first_facts: DList[Fact], second_facts: DList[Fact], outputPath: String): ScoobiAction[Unit] = {
     ScoobiAction.scoobiJob({ implicit sc: ScoobiConfiguration =>
-      val (first_errs, first_facts) = byflag(PartitionFactThriftStorageV1.PartitionedFactThriftLoader(List(input1)).loadScoobi, true)
-      val (second_errs, second_facts) = byflag(PartitionFactThriftStorageV1.PartitionedFactThriftLoader(List(input2)).loadScoobi, false)
 
-      val errors = first_errs ++ second_errs
-      val facts = first_facts ++ second_facts
+      val facts = first_facts.map((true, _)) ++ second_facts.map((false, _))
 
       val grp = facts.groupBy({ case (flag, fact) => (fact.entity, fact.featureId.toString, fact.date.int, fact.time.seconds, fact.value.stringValue) })
 
@@ -31,28 +54,12 @@ object FactDiff {
       })
 
       val out: DList[String] = diff.map({
-        case (true, fact) :: Nil  => s"Fact '${fact}' does not exist in ${input2}"
-        case (false, fact) :: Nil => s"Fact '${fact}' does not exist in ${input1}"
+        case (true, fact) :: Nil  => s"Fact '${fact}' does not exist in input2"
+        case (false, fact) :: Nil => s"Fact '${fact}' does not exist in input1"
         case g                    => s"Found duplicates - '${g}'"
       })
 
-      val error_out: DList[String] = errors.map({
-        case (true, e)  => s"${e.message} - ${input1}"
-        case (false, e) => s"${e.message} - ${input2}"
-      })
-
-      persist(error_out.toTextFile(errorPath, overwrite = true), out.toTextFile(outputPath, overwrite = true))
+      persist(out.toTextFile(outputPath, overwrite = true))
     })
   }
-
-  def byflag(dlist: DList[ParseError \/ Fact], flag: Boolean): (DList[(Boolean, ParseError)], DList[(Boolean, Fact)]) = {
-    val errs = dlist.collect {
-      case -\/(e) => (flag, e)
-    }
-
-    val facts = dlist.collect {
-      case \/-(f) => (flag, f)
-    }
-    (errs, facts)
-  }
 }

ivory-validate/src/test/scala/com/ambiata/ivory/validate/FactDiffSpec.scala

@@ -36,17 +36,14 @@ class FactDiffSpec extends HadoopSpecification with SimpleJobs with FileMatchers
     val input1 = directory + "/1"
     val input2 = directory + "/2"
     val output = directory + "/out"
-    val errors = directory + "/errors"
 
     persist(PartitionFactThriftStorageV1.PartitionedFactThriftStorer(input1, None).storeScoobi(facts1),
             PartitionFactThriftStorageV1.PartitionedFactThriftStorer(input2, None).storeScoobi(facts2))
 
-    FactDiff.scoobiJob(input1, input2, output, errors).run(sc) must beOk
+    FactDiff.partitionFacts(input1, input2, output).run(sc) must beOk
 
     val out = fromTextFile(output).run.toList
     out must have size(6)
-
-    Hdfs.readWith(new Path(errors), is => Streams.read(is)).run(sc) must beOkValue("")
   }
 
   "FactDiff finds no difference" >> { implicit sc: ScoobiConfiguration =>
@@ -59,15 +56,13 @@ class FactDiffSpec extends HadoopSpecification with SimpleJobs with FileMatchers
     val input1 = directory + "/1"
     val input2 = directory + "/2"
     val output = directory + "/out"
-    val errors = directory + "/errors"
 
     persist(PartitionFactThriftStorageV1.PartitionedFactThriftStorer(input1, None).storeScoobi(facts1),
             PartitionFactThriftStorageV1.PartitionedFactThriftStorer(input2, None).storeScoobi(facts1))
 
-    FactDiff.scoobiJob(input1, input2, output, errors).run(sc) must beOk
+    FactDiff.partitionFacts(input1, input2, output).run(sc) must beOk
 
     Hdfs.readWith(new Path(output), is => Streams.read(is)).run(sc) must beOkValue("")
-    Hdfs.readWith(new Path(errors), is => Streams.read(is)).run(sc) must beOkValue("")
   }
 
 }