修改rdd

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

@@ -55,11 +55,11 @@ object Data2FFM {
 //      val yesteday_have_seq = GmeiConfig.getMinusNDate(5)
       val esmm_data = sc.sql(
         s"""
-           |select device_id,y,z,stat_date,ucity_id,cid_id,clevel1_id,ccity_name
-           |from esmm_train_data limit 6
+           |select device_id,y,z,stat_date,ucity_id,cid_id,clevel1_id,ccity_name from esmm_train_data limit 6
          """.stripMargin
       ).na.drop()
-      esmm_data.show(6)
+      esmm_data.show(8)
+      println(esmm_data.count())
       val column_list = esmm_data.columns
 
       val max_stat_date = sc.sql(
@@ -68,9 +68,7 @@ object Data2FFM {
          """.stripMargin
       )
       println("------------------------")
-      val max_stat_date_str = max_stat_date.collect().map(s =>
-        s(0).toString
-      ).head
+      val max_stat_date_str = max_stat_date.collect().map(s => s(0).toString).head
 
       println(max_stat_date_str)
 
@@ -81,17 +79,16 @@ object Data2FFM {
         column_number(i) = esmm_data.select(i).distinct().collect().map(x => x(0).toString)
       }
 
-
       val a  = column_number("device_id").toList
       println(a)
       println("dict")
       val rdd = esmm_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))
+          x(4).toString,x(5).toString,x(6).toString, x(7).toString))
       rdd.persist()
 
       val b = rdd.map(x => x._1).collect().toList
+      println("rdd")
       println(b)
 
       import sc.implicits._