repair bug

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

 package com.gmei
 
 import java.io.Serializable
+import org.apache.spark.sql.functions.udf
 
 import com.gmei.WeafareStat.{defaultParams, parser}
 import org.apache.spark.sql.{SaveMode, TiContext}
@@ -61,6 +62,15 @@ object app_list {
       val partition_date = param.date.replace("-","")
       println(partition_date)
 
+      //自定义udf函数，增加dataframe 列
+      val code = (arg: String) => {
+        if (arg.getClass.getName == "java.lang.String") partition_date else 0
+      }
+
+      val addCol = udf(code)
+      //以上为udf函数
+
+
       //获取策略命中用户device_id
       val app_list = sc.sql(
         s"""
@@ -72,10 +82,10 @@ object app_list {
       )
       //app_list.show()
       import sc.implicits._
-      val rdd = app_list.rdd.map(x =>(x(0).toString,x(1).toString))
-        .filter(x => x._2.contains("新氧美容")).map(x => x._1).collect().toList.toDF()
-      rdd.show()
-      rdd.createOrReplaceTempView("device_id")
+      val rdd_df = app_list.rdd.map(x =>(x(0).toString,x(1).toString))
+        .filter(x => x._2.contains("新氧美容")).map(x => x._1).collect().toList.toDF("device_id")
+      rdd_df.show()
+      rdd_df.createOrReplaceTempView("device_id")
 
       val temp = sc.sql(
         s"""
@@ -83,6 +93,7 @@ object app_list {
            |from device_id
          """.stripMargin
       )
+      temp.withColumn("stat_date",addCol(temp("device_id")))
 
       temp.show()