更改时间

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

@@ -8,6 +8,10 @@ import org.apache.spark.sql.SaveMode
 import org.apache.log4j.{Level, Logger}
 import scopt.OptionParser
 import com.gmei.lib.AbstractParams
+//import com.sun.deploy.uitoolkit.Window
+
+import org.apache.spark.sql.expressions.Window
+import org.apache.spark.sql.functions.row_number
 
 object data_feed_exposure_precise {
 
@@ -662,7 +666,7 @@ object tag_value {
       val t2=0.5
       val t3=1.0
 
-      //tag_id到权重映射
+      //level_id到权重映射
       val level1_id=sc.sql(
         s"""
            |select distinct(level1_id) as level_id,'${t1}' as value
@@ -694,7 +698,7 @@ object tag_value {
       level3_id.createOrReplaceTempView("level3_id")
 
 
-      val tag_value=sc.sql(
+      val level_value=sc.sql(
         s"""
            |select *
            |from level1_id
@@ -709,10 +713,59 @@ object tag_value {
 //      tag_value.show(300)
 
       println("开始写入")
-      GmeiConfig.writeToJDBCTable("jdbc:mysql://172.16.40.158:4000/jerry_prod?user=root&password=3SYz54LS9#^9sBvC&rewriteBatchedStatements=true",tag_value, table="tagId_value",SaveMode.Overwrite)
+      GmeiConfig.writeToJDBCTable("jdbc:mysql://172.16.40.158:4000/jerry_prod?user=root&password=3SYz54LS9#^9sBvC&rewriteBatchedStatements=true",level_value, table="tagId_value",SaveMode.Overwrite)
       println("写入完成")
 
 
+      //level_id到index映射
+
+      val level_index_temp=sc.sql(
+        s"""
+           |select a.tag_id,b.level1_id,b.level2_id,b.level3_id
+           |from online.tl_hdfs_diary_tags_view a
+           |left join online.bl_tag_hierarchy_detail b
+           |on a.tag_id = b.id
+           |where a.partition_date = '${partition_date}'
+           |and b.partition_date = '${partition_date}'
+       """.stripMargin
+      )
+      level_index_temp.createOrReplaceTempView("level_index_temp")
+
+      val temp1=sc.sql(
+        s"""
+           |select tag_id,level1_id as level_id
+           |from level_index_temp
+           |union all
+           |select tag_id,level2_id as level_id
+           |from level_index_temp
+           |union all
+           |select tag_id,level3_id as level_id
+           |from level_index_temp
+       """.stripMargin
+      )
+      temp1.createOrReplaceTempView("temp1")
+
+
+
+      val temp2=sc.sql(
+        s"""
+           |select concat_ws(',',tag_id,level_id) as tag_level from temp1
+       """.stripMargin
+      )
+      temp2.show()
+//      temp2.createOrReplaceTempView("temp2")
+
+      val w = Window.orderBy("tag_level")
+      val result = temp2.select("tag_level").distinct().withColumn("index", row_number().over(w))
+      result.show()
+
+
+
+
+
+
+
+
     }