Get inactive customers using left outer join between orders and customers.

spark-scala-get-inactive-customers.scala

/*
spark-shell --master yarn \
  --conf spark.ui.port=12345 \
  --num-executors 1 \
  --executor-cores 1 \
  --executor-memory 2G
*/

import scala.io.Source

val ordersRaw = Source.fromFile("/data/retail_db/orders/part-00000").getLines.toList
val orders = sc.parallelize(ordersRaw)

val customersRaw = Source.fromFile("/data/retail_db/customers/part-00000").getLines.toList
val customers = sc.parallelize(customersRaw)

val ordersMap = orders.
  map(order => (order.split(",")(2).toInt, 1))
val customersMap = customers.
  map(c => (c.split(",")(0).toInt, (c.split(",")(2), c.split(",")(1))))
val customersLeftOuterJoinOrders = customersMap.leftOuterJoin(ordersMap)
val inactiveCustomersSorted = customersLeftOuterJoinOrders.
  filter(t => t._2._2 == None).
  map(rec => rec._2).
  sortByKey()
inactiveCustomersSorted.
  map(rec => rec._1._1 + ", " + rec._1._2).
  saveAsTextFile("/user/dgadiraju/solutions/solutions02/inactive_customers")

Hi Sir..
I have one query.. don't we have to use distinct before sorting the data as there are duplicates names?

@dgadiraju I also think that we should first get the list of unique customers from orders table(as one customer can have multiple orders) and then find the customers from customer rdd which are not in that unique customer list.