spark-core-intersection-01-customer-ids.scala

val orderCustomers1 = sc.textFile("/public/retail_db/orders").
  filter(o => o.split(",")(1).contains("2013-08")).
  map(o => o.split(",")(2).toInt)

val orderCustomers2 = sc.textFile("/public/retail_db/orders").
  filter(o => o.split(",")(1).contains("2013-09")).
  map(o => o.split(",")(2).toInt)

orderCustomers1.count
orderCustomers2.count

orderCustomers1.
  intersection(orderCustomers2).
  take(10).
  foreach(println)

orderCustomers1.
  intersection(orderCustomers2).
  count

spark-core-intersection-02-customer-details.scala

val orderCustomers1 = sc.textFile("/public/retail_db/orders").
  filter(o => o.split(",")(1).contains("2013-08")).
  map(o => o.split(",")(2).toInt)

val orderCustomers2 = sc.textFile("/public/retail_db/orders").
  filter(o => o.split(",")(1).contains("2013-09")).
  map(o => o.split(",")(2).toInt)

val customers = sc.textFile("/public/retail_db/customers").
  map(c => (c.split(",")(0).toInt, c))

orderCustomers1.
  intersection(orderCustomers2).
  map(c => (c, 1)).
  join(customers).
  map(c => c._2._2).
  take(10).
  foreach(println)