更改时间

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

@@ -4,13 +4,15 @@ import java.io.Serializable
 
 import breeze.linalg.split
 import com.gmei.WeafareStat.{defaultParams, parser}
-import org.apache.spark.sql.SaveMode
+import org.apache.spark.sql.{Row, SaveMode}
 import org.apache.log4j.{Level, Logger}
 import scopt.OptionParser
 import com.gmei.lib.AbstractParams
+import com.sun.jdi.IntegerType
+import org.apache.spark
 import org.apache.spark.sql.expressions.Window
 import org.apache.spark.sql.functions.row_number
-import org.apache.spark.sql.types.DoubleType
+import org.apache.spark.sql.types.{DoubleType, StructField, StructType}
 
 
 object data_feed_exposure_precise {
@@ -745,6 +747,12 @@ object tag_value {
        """.stripMargin
       )
 //      temp1.createOrReplaceTempView("temp1")
+//      import  implicit_
+
+      val result = temp1.select("level_id").distinct().rdd.map{x => x.toString().substring(1,x.toString().length - 1)}.zipWithIndex().toDF("level_id","index")
+      val test=result.select(result.col("level_id").cast(DoubleType).as("level_id"),result.col("index").cast(DoubleType).as("index"))
+//      val resDf = spark.createDataFrame(rowRdd)
+
 
 
 
@@ -756,9 +764,10 @@ object tag_value {
 //      temp2.show()
 //      temp2.createOrReplaceTempView("temp2")
 
-      val w = Window.orderBy("level_id")
-      val result = temp1.select("level_id").distinct().withColumn("index", row_number().over(w))
-      result.show()
+
+//      val w = Window.orderBy("level_id")
+//      val result = temp1.select("level_id").distinct().withColumn("index", row_number().over(w))
+//      result.show()
 
 //      val tag_level_index=result.rdd.map(row=>(row(0).toString,row(1).toString)).map(row=>(row._1.split(",").head,row._1.split(",")(1),row._2)).toDF("tag_id","level_id","index")
 
@@ -768,7 +777,7 @@ object tag_value {
 //      val test=tag_level_index.select(tag_level_index.col("tag_id").cast(DoubleType).as("tag_id"),tag_level_index.col("level_id").cast(DoubleType).as("level_id"),tag_level_index.col("index").cast(DoubleType).as("index"))
 
       println("开始写入")
-      GmeiConfig.writeToJDBCTable("jdbc:mysql://172.16.40.158:4000/jerry_prod?user=root&password=3SYz54LS9#^9sBvC&rewriteBatchedStatements=true",result, table="tag_level_index",SaveMode.Overwrite)
+      GmeiConfig.writeToJDBCTable("jdbc:mysql://172.16.40.158:4000/jerry_prod?user=root&password=3SYz54LS9#^9sBvC&rewriteBatchedStatements=true",test, table="tag_level_index",SaveMode.Overwrite)
       println("写入完成")