spark-core-compute-daily-revenue.scala

val orders = sc.textFile("/public/retail_db/orders")
val orderItems = sc.textFile("/public/retail_db/order_items")

val ordersCompleted = orders.
  filter(o => List("COMPLETE", "CLOSED").contains(o.split(",")(3)))

val ordersCompletedMap = ordersCompleted.
  map(o =>(o.split(",")(0).toInt, o.split(",")(1)))

val orderItemsMap = orderItems.
  map(o =>(o.split(",")(1).toInt, o.split(",")(4).toFloat))
  
val ordersJoin = ordersCompletedMap.
  join(orderItemsMap)
  
val ordersJoinMap = ordersJoin.
  map(o => o._2)
  
val dailyRevenue = ordersJoinMap.
  reduceByKey((agg, rev) => agg + rev)
  
val dailyRevenueSorted = dailyRevenue.
  sortByKey()

val dailyRevenueSortedMap = dailyRevenueSorted.
  map(o => o.productIterator.mkString(","))
  
dailyRevenueSortedMap.
  saveAsTextFile("/user/training/daily_revenue")