1 package class3
 2  
 3 import org.apache.spark.SparkContext._
 4 import org.apache.spark.{SparkConf, SparkContext}
 5  
 6 object Join{
 7   def main(args: Array[String]) {
 8     if (args.length == 0) {
 9       System.err.println("Usage: Join <file1> <file2>")
10       System.exit(1)
 
11     }
12  
13     val conf = new SparkConf().setAppName("Join").setMaster("local")
14     val sc = new SparkContext(conf)
15  
16     val format = new java.text.SimpleDateFormat("yyyy-MM-dd")
17     case class Register (d: java.util.Date, uuid: String, cust_id: String, lat: Float,lng: Float)
 
18     case class Click (d: java.util.Date, uuid: String, landing_page: Int)
19     val reg = sc.textFile(args(0)).map(_.split("\t")).map(r => (r(1), Register(format.parse(r(0)), r(1), r(2), r(3).toFloat, r(4).toFloat)))
20 val clk = sc.textFile(args(1)).map(_.split("\t")).map(c => (c(1), Click(format.parse(c(0)), c(1), c(2).trim.toInt)))
 
21     reg.join(clk).take(2).foreach(println)
22  
23     sc.stop()
24   }
25 }