example of LAG type function in the scalding Fields API (similar for typed)

gistfile1.scala

groupBy('source) {
  _.sortBy('links)
    .reverse
    .mapStream[(String,Int), (String, Int, Int, Int)]
      (('destination, 'links) -> ('destination, 'links, 'rank, 'gap)) { destLinks =>
        destLinks.scanLeft(None: Option[(String, Int, Int, Int)]) {
          (prevRowOut: Option[(String,Int,Int,Int)], thisRow: (String, Int))  =>
          val (dest, links) = thisRow
          prevRowOut match {
            case None => Some((dest, links, 1, 0)) // rank 1, gap 0 -- not exactly what you wanted...
            case Some((prevDest, prevLinks, prevRank, prevGap)) =>
              Some(dest, links, prevRank + 1, prevLinks - links)
          }
        }
        .collect { case Some(x) => x } // drop the initial None, and unwrap 
}