Scalding benchmark

quickAndDirtyScaldingBenchmark.scala

def getTiming(ebScaldVariant:String, idx: Int, runName: String, tp: TypedPipe[_]) = {
    val ms = System.currentTimeMillis
    val cntrs = execute(tp.filter(_ => false).toIterableExecution.getAndResetCounters.map(_._2))
    val after = System.currentTimeMillis
    val cpuMS = cntrs.get(StatKey("CPU_MILLISECONDS", "org.apache.hadoop.mapreduce.TaskCounter")).get
    val gcMS = cntrs.get(StatKey("GC_TIME_MILLIS", "org.apache.hadoop.mapreduce.TaskCounter")).get
    val millisMap = cntrs.get(StatKey("MILLIS_MAPS", "org.apache.hadoop.mapreduce.JobCounter")).get
    (ebScaldVariant, idx, runName, cpuMS, gcMS, millisMap, after - ms)
}


val r = for {
  indx <- (0 until 1).toList
  (runName, runTP) <- List(("old", oldTP), ("new", newTP))
} yield getTiming("resultsData", indx, runName, runTP)

useLocalMode

TypedPipe.from(r).save(TypedTsv("resultsData.tsv"))

println("all done")