Merge branch 'master' of git.wanmeizhensuo.com:ML/ffm-baseline

esmm 测试项目

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

@@ -558,85 +558,85 @@ object alpha_ctr {
       //println(param.date)
       val partition_date = stat_date.replace("-","")
 
-//      val click_count_recommend = sc.sql(
-//        s"""
-//           |select '${stat_date}' as stat_date,count(*) as click_count_recommend
-//           |from bl.bl_alpha_et_mg_maidianlog_inc_d
-//           |where params['tab_name']='recommend'
-//           |and params['page_name']='home'
-//           |and type='on_click_feed_topic_card'
-//           |and partition_day='${partition_date}'
-//       """.stripMargin
-//      )
-//      click_count_recommend.show()
-//
-//      val click_count_focus = sc.sql(
-//        s"""
-//           |select '${stat_date}' as stat_date,count(*) as click_count_focus
-//           |from bl.bl_alpha_et_mg_maidianlog_inc_d
-//           |where params['tab_name']='focus'
-//           |and params['page_name']='home'
-//           |and type='on_click_feed_topic_card'
-//           |and partition_day='${partition_date}'
-//       """.stripMargin
-//      )
-//      click_count_focus.show()
-//
-//
-//      def parse_json(str:String): Int ={
-//        var t = List[Map[String, Any]]()
-//        val result = JSON.parseFull(str)
-//        result match {
-//          case Some(b: List[Map[String, Any]]) => t = t ++ b
-//          case None => println("Parsing failed")
-//          case other => println("Unknown data structure: " + other)
-//        }
-//        t.size
-//
-//      }
-//
-//      val expoure_cards=sc.sql(
-//        s"""
-//           |select params['exposure_cards'] as exposure_cards
-//           |from bl.bl_alpha_et_mg_maidianlog_inc_d
-//           |where params['tab_name'] = 'recommend'
-//           |and params['page_name'] = 'home'
-//           |and type = 'page_precise_exposure'
-//           |and partition_day='${partition_date}'
-//       """.stripMargin
-//      )
-//      val a =expoure_cards.rdd.map(row => row(0).toString).map(row=>parse_json(row)).collect().sum
-//      val result1=List((stat_date,a))
-//      val df1 = sc.createDataFrame(result1).toDF("stat_date","expoure_count_recommend")
-//
-//      val expoure_cards2=sc.sql(
-//        s"""
-//           |select params['exposure_cards'] as exposure_cards
-//           |from bl.bl_alpha_et_mg_maidianlog_inc_d
-//           |where params['tab_name'] = 'focus'
-//           |and params['page_name'] = 'home'
-//           |and type = 'page_precise_exposure'
-//           |and partition_day='${partition_date}'
-//       """.stripMargin
-//      )
-//      val b =expoure_cards2.rdd.map(row => row(0).toString).map(row=>parse_json(row)).collect().sum
-//      val result2=List((stat_date,b))
-//      val df2 = sc.createDataFrame(result2).toDF("stat_date","expoure_count_focus")
-//
-//
-//
-//      val result=click_count_recommend.join(click_count_focus,"stat_date")
-//        .join(df1,"stat_date")
-//        .join(df2,"stat_date")
-//
-//
-//
-//      GmeiConfig.writeToJDBCTable(result, "alpha_ctr", SaveMode.Append)
+      val click_count_recommend = sc.sql(
+        s"""
+           |select '${stat_date}' as stat_date,count(*) as click_count_recommend
+           |from bl.bl_alpha_et_mg_maidianlog_inc_d
+           |where params['tab_name']='recommend'
+           |and params['page_name']='home'
+           |and type='on_click_feed_topic_card'
+           |and partition_day='${partition_date}'
+       """.stripMargin
+      )
+      click_count_recommend.show()
+
+      val click_count_focus = sc.sql(
+        s"""
+           |select '${stat_date}' as stat_date,count(*) as click_count_focus
+           |from bl.bl_alpha_et_mg_maidianlog_inc_d
+           |where params['tab_name']='focus'
+           |and params['page_name']='home'
+           |and type='on_click_feed_topic_card'
+           |and partition_day='${partition_date}'
+       """.stripMargin
+      )
+      click_count_focus.show()
+
+
+      def parse_json(str:String): Int ={
+        var t = List[Map[String, Any]]()
+        val result = JSON.parseFull(str)
+        result match {
+          case Some(b: List[Map[String, Any]]) => t = t ++ b
+          case None => println("Parsing failed")
+          case other => println("Unknown data structure: " + other)
+        }
+        t.size
+
+      }
+
+      val expoure_cards=sc.sql(
+        s"""
+           |select params['exposure_cards'] as exposure_cards
+           |from bl.bl_alpha_et_mg_maidianlog_inc_d
+           |where params['tab_name'] = 'recommend'
+           |and params['page_name'] = 'home'
+           |and type = 'page_precise_exposure'
+           |and partition_day='${partition_date}'
+       """.stripMargin
+      )
+      val a =expoure_cards.rdd.map(row => row(0).toString).map(row=>parse_json(row)).collect().sum
+      val result1=List((stat_date,a))
+      val df1 = sc.createDataFrame(result1).toDF("stat_date","expoure_count_recommend")
+
+      val expoure_cards2=sc.sql(
+        s"""
+           |select params['exposure_cards'] as exposure_cards
+           |from bl.bl_alpha_et_mg_maidianlog_inc_d
+           |where params['tab_name'] = 'focus'
+           |and params['page_name'] = 'home'
+           |and type = 'page_precise_exposure'
+           |and partition_day='${partition_date}'
+       """.stripMargin
+      )
+      val b =expoure_cards2.rdd.map(row => row(0).toString).map(row=>parse_json(row)).collect().sum
+      val result2=List((stat_date,b))
+      val df2 = sc.createDataFrame(result2).toDF("stat_date","expoure_count_focus")
+
+
+
+      val result=click_count_recommend.join(click_count_focus,"stat_date")
+        .join(df1,"stat_date")
+        .join(df2,"stat_date")
+
+
+
+      GmeiConfig.writeToJDBCTable(result, "alpha_ctr", SaveMode.Append)
 
 
       val device_duration_avge = sc.sql(
         s"""
-           |SELECT '${stat_date}' as stat_date,sum(a.time_all)/count(a.device_id) as averge_time from (select device_id,sum(params['duration']) as time_all
+           |SELECT '${stat_date}' as stat_date,sum(a.time_all)/count(a.device_id) as device_duration_avge,count(distinct(a.device_id)) as device_num from (select device_id,sum(params['duration']) as time_all
            |from bl.bl_alpha_et_mg_maidianlog_inc_d
            |where type='on_app_session_over'
            |and partition_day='${partition_date}'
@@ -647,10 +647,10 @@ object alpha_ctr {
 
       val duration_everytime_avge=sc.sql(
         s"""
-           |SELECT '${stat_date}' as stat_date,sum(a.time_duration)/count(a.device_id) as averge_time from (select device_id,params['duration'] as time_duration
+           |SELECT '${stat_date}' as stat_date,sum(a.time_duration)/count(a.device_id) as duration_everytime_avge from (select device_id,params['duration'] as time_duration
            |from bl.bl_alpha_et_mg_maidianlog_inc_d
            |where type='on_app_session_over'
-           |and partition_day='20190404') a
+           |and partition_day='${partition_date}') a
        """.stripMargin
       )