增加rdd持久化和分区

eda/feededa/src/main/scala/com/gmei/Data2FFM.scala

@@ -57,8 +57,8 @@ object Data2FFM {
         s"""
            |select device_id,y,z,stat_date,ucity_id,cid_id,clevel1_id,ccity_name from esmm_train_data
          """.stripMargin
-      ).na.drop()
-      val column_list = esmm_data.columns
+      ).repartition(200).na.drop()
+      val column_list = esmm_data.columns.filter(x => x != "y" && x != "z")
       val max_stat_date = sc.sql(
         s"""
            |select max(stat_date) from esmm_train_data
@@ -70,19 +70,18 @@ object Data2FFM {
       println(max_stat_date_str)
 
       println(column_list.slice(0,2).toList)
-
+      esmm_data.persist()
       val column_number = scala.collection.mutable.Map[String,Array[String]]()
       for (i <- column_list){
-        column_number(i) = esmm_data.select(i).distinct().collect().map(x => x(0).toString)
+        column_number(i) = esmm_data.select(i).collect().map(x => x(0).toString).distinct
       }
-
+      esmm_data.unpersist()
       println("dict")
-      val rdd = esmm_data.rdd.repartition(200)
+      val rdd = esmm_data.rdd
         .map(x => (x(0).toString,x(1).toString,x(2).toString,x(3).toString,
           x(4).toString,x(5).toString,x(6).toString, x(7).toString))
       rdd.persist()
 
-
       import sc.implicits._
       val train = rdd.filter(x => x._4 != max_stat_date_str)
         .map(x => (x._1,x._2,x._3,column_number("device_id").indexOf(x._1),
@@ -109,6 +108,7 @@ object Data2FFM {
         .map(x => (x._1,x._2+","+x._3+","+x._4+","+x._5,x._6,x._7,x._8)).toDF("number","data","device_id","city_id","cid")
       println("test")
       test.show(6)
+      rdd.unpersist()
       GmeiConfig.writeToJDBCTable(jdbcuri, test, "esmm_data2ffm_cv", SaveMode.Overwrite)
 
 
@@ -117,7 +117,7 @@ object Data2FFM {
            |select device_id,y,z,stat_date,ucity_id,cid_id,clevel1_id,ccity_name
            |from esmm_pre_data
         """.stripMargin
-      ).na.drop()
+      ).repartition(200).na.drop()
 
       val esmm_pre_cids = esmm_pre_data.select("cid_id").distinct().collect().map(
         s => s(0).toString
@@ -125,19 +125,16 @@ object Data2FFM {
       val esmm_pre_city = esmm_pre_data.select("ucity_id").distinct().collect().map(
         s => s(0).toString
       )
-      val esmm_pre_device = esmm_pre_data.select("device_id").distinct().collect().map(
-        s => s(0).toString
-      )
+
       val esmm_join_cids = esmm_pre_cids.intersect(column_number("cid_id"))
       val esmm_join_city = esmm_pre_city.intersect(column_number("ucity_id"))
-      val esmm_join_device = esmm_pre_device.intersect(column_number("device_id"))
 
       val rdd_pre = esmm_pre_data.rdd.repartition(200)
         .map(x => (x(0).toString,x(1).toString,x(2).toString,x(3).toString,
           x(4).toString,x(5).toString,x(6).toString,
           x(7).toString)).filter(x => esmm_join_cids.indexOf(x._6) != -1)
         .filter(x => esmm_join_city.indexOf(x._5) != -1)
-        .filter(x => esmm_join_device.indexOf(x._1) != -1)
+
       val pre = rdd_pre.map(x => (x._1,x._2,x._3,column_number("device_id").indexOf(x._1),
         column_number("stat_date").indexOf(x._4), column_number("ucity_id").indexOf(x._5),
         column_number("cid_id").indexOf(x._6), column_number("clevel1_id").indexOf(x._7),