addd city_id

eda/node2vec/src/main/scala/com/gmei/Main.scala

@@ -190,24 +190,25 @@ object Main {
 
       val device_id = sc.sql(
         s"""
-           |select a.device_id device_id,b.similarity_cid similarity_cid from
-           |(select device_id,first(cid) as cid from data_feed_click
+           |select a.device_id device_id,a.city_id city_id ,b.similarity_cid similarity_cid from
+           |(select device_id,city_id,first(cid) as cid from data_feed_click
            |where cid in (select cid from nd_cid_similarity_matrix)
-           |group by device_id) a left join
+           |group by device_id order by time) a left join
            |nd_cid_similarity_matrix b
            |on a.cid = b.cid
            |where b.similarity_cid is not null
      """.stripMargin
       )
+      device_id.na.fill(Map("city_id" -> "beijing"))
       device_id.show()
 
       val device_queue = device_id.rdd.map {item =>
-        val parts = (item.getAs[String](fieldName = "device_id"),item.getAs[String](fieldName = "similarity_cid"))
+        val parts = (item.getAs[String](fieldName = "device_id"),item.getAs[String](fieldName = "city_id"),item.getAs[String](fieldName = "similarity_cid"))
 
         Try {
-          (parts._1, Try(parts._2.toString.replace("diary|","")).getOrElse(null))
+          (parts._1,Try(parts._2.toString.replace("worldwide","beijing")),Try(parts._3.toString.replace("diary|","")).getOrElse(null))
         }.getOrElse(null)
-      }.filter(_!=null).toDF("device_id","similarity_cid")
+      }.filter(_!=null).toDF("device_id","city_id","similarity_cid")
 
       device_queue.take(20).foreach(println)
       GmeiConfig.writeToJDBCTable(device_queue, table="nd_device_cid_similarity_matrix", SaveMode.Overwrite)