Doing basic set operations in scalding

set_join.scala

// how do do set operations in scalding:

case class DSet[T](toPipe: TypedPipe[T]) {
  // returns the items in that that are in toPipe (not-de-duplicated
  def contains(that: TypedPipe[T])(implicit ord: Ordering[T]): TypedPipe[T] =
    toPipe.asKeys.sum // reduce all the values to a single value on the set
      .join(that.asKeys.size)
      .flatMap { case (t, (_, sz)) => Iterator.fill(sz)(t) }

  // which items in that are not in toPipe, the union
  // of contains and notContains is the original that.
  def notContains(that: TypedPipe[T])(implicit ord: Ordering[T]): TypedPipe[T] =
    toPipe.asKeys.sum
      .rightJoin(that.asKeys.size)
      .flatMap {
        case (t, (None, Some(sz))) => Iterator.fill(sz)(t)
        case (_, (Some(_), _)) => Iterator.empty // exists in toPipep       
      }
}