Examples for Optimize Scalding Jobs

count_uniques.scala

// 2 m/r jobs :-(
.unique('item_id_from, 'item_id_to, 'user_id)            // 1st m/r
.groupBy('item_id_from, 'item_id_to) { _.size('count) }  // 2nd m/r

// 1 m/r job but more code
.map('user_id -> 'user_id) { id: String => Set(id) }
.groupBy('item_id_from, 'item_id_to) {
  _.sum[Set[String}]('user_id)
}
.map('user_id -> 'count) { ids: Set[String] => ids.size }

// 1 m/r job  :-)
.groupBy('item_id_from, 'item_id_to) {
  _.mapReduceMap[String, Set[String], Int]('user_id -> 'count)(Set(_))(_ ++ _)(_.size)
}

// 1m/r job for big numbers (HLL)
.groupBy('item_id_from, 'item_id_to) {
  import util.utf8
  _.approximateUniqueCount[String]('user_id -> 'count)
}
.map('count -> 'count) { count: Double => count.toLong }

parquet_projection.scala

def readTaps = ParquetAvroSource.project[Tap](tapsInput, 
  Projection[Tap]("campaign_id", "action_id", "header.device.storage_key")).read