temp_analysis.scala

package com.gmei

import java.io.Serializable

import com.gmei.WeafareStat.{defaultParams, parser}
import org.apache.spark.sql.{SaveMode, TiContext}
import org.apache.log4j.{Level, Logger}
import scopt.OptionParser
import com.gmei.lib.AbstractParams
import java.io._

object temp_analysis {

  Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
  Logger.getLogger("org.apache.eclipse.jetty.server").setLevel(Level.OFF)

  case class Params(env: String = "dev",
                    date: String = "2018-08-01"
                   ) extends AbstractParams[Params] with Serializable

  val defaultParams = Params()

  val parser = new OptionParser[Params]("Feed_EDA") {
    head("WeafareStat")
    opt[String]("env")
      .text(s"the databases environment you used")
      .action((x, c) => c.copy(env = x))
    opt[String] ("date")
      .text(s"the date you used")
      .action((x,c) => c.copy(date = x))
    note(
      """
        |For example, the following command runs this app on a tidb dataset:
        |
        | spark-submit --class com.gmei.WeafareStat ./target/scala-2.11/feededa-assembly-0.1.jar \
      """.stripMargin +
        s"|   --env ${defaultParams.env}"
    )
  }

  def main(args: Array[String]): Unit = {
    parser.parse(args, defaultParams).map { param =>
      GmeiConfig.setup(param.env)
      val spark_env = GmeiConfig.getSparkSession()
      val sc = spark_env._2

      val ti = new TiContext(sc)
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "diary_video")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "data_feed_click")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "blacklist")
      ti.tidbMapTable(dbName = "jerry_test", tableName = "bl_device_list")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "data_feed_exposure")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "merge_queue_table")


      import sc.implicits._
      val stat_date = GmeiConfig.getMinusNDate(1)
      //println(param.date)
      val partition_date = stat_date.replace("-","")

      val agency_id = sc.sql(
        s"""
           |SELECT DISTINCT(cl_id) as device_id
           |FROM online.ml_hospital_spam_pv_day
           |WHERE partition_date >= '20180402'
           |AND partition_date <= '20181203'
           |AND pv_ratio >= 0.95
           |UNION ALL
           |SELECT DISTINCT(cl_id) as device_id
           |FROM online.ml_hospital_spam_pv_month
           |WHERE partition_date >= '20171101'
           |AND partition_date <= '20181203'
           |AND pv_ratio >= 0.95
         """.stripMargin
      )
      agency_id.createOrReplaceTempView("agency_id")


//      //每日新用户
//      val device_id_newUser = sc.sql(
//        s"""
//           |select distinct(device_id) as device_id
//           |from online.ml_device_day_active_status
//           |where active_type != '4'
//           |and first_channel_source_type not in ('yqxiu1','yqxiu2','yqxiu3','yqxiu4','yqxiu5','mxyc1','mxyc2','mxyc3'
//           |        ,'wanpu','jinshan','jx','maimai','zhuoyi','huatian','suopingjingling','mocha','mizhe','meika','lamabang'
//           |        ,'js-az1','js-az2','js-az3','js-az4','js-az5','jfq-az1','jfq-az2','jfq-az3','jfq-az4','jfq-az5','toufang1'
//           |        ,'toufang2','toufang3','toufang4','toufang5','toufang6','TF-toufang1','TF-toufang2','TF-toufang3','TF-toufang4'
//           |        ,'TF-toufang5','tf-toufang1','tf-toufang2','tf-toufang3','tf-toufang4','tf-toufang5','benzhan','promotion_aso100'
//           |        ,'promotion_qianka','promotion_xiaoyu','promotion_dianru','promotion_malioaso','promotion_malioaso-shequ'
//           |        ,'promotion_shike','promotion_julang_jl03')
//           |and partition_date ='${partition_date}'
//         """.stripMargin
//      )
//      device_id_newUser.createOrReplaceTempView("device_id_new")

      val blacklist_id = sc.sql(
        s"""
           |SELECT device_id
           |from blacklist
         """.stripMargin
      )
      blacklist_id.createOrReplaceTempView("blacklist_id")

      val final_id = sc.sql(
        s"""
           |select device_id
           |from agency_id
           |UNION ALL
           |select device_id
           |from blacklist_id
         """.stripMargin
      )
      final_id.createOrReplaceTempView("final_id")


      val diary_clk_all = sc.sql(
        s"""
           |select ov.partition_date,count(ov.cl_id) as clk_num,count(distinct(ov.cl_id)),count(ov.cl_id)/count(distinct(ov.cl_id))
           |from online.tl_hdfs_maidian_view ov left join final_id
           |on ov.cl_id = final_id.device_id
           |where ov.action = "page_view"
           |and params['page_name']="diary_detail"
           |and ov.cl_id != "NULL"
           |and ov.partition_date >='20181201'
           |and final_id.device_id is  null
           |group by ov.partition_date
           |order by ov.partition_date
       """.stripMargin
      )
      diary_clk_all.show(80)


//日记本点击
      val referrer=List("about_me_message_list","all_case_service_comment","all_cases","diary_detail","diary_list"
        ,"diary_listof_related_service","answer_detail","community_home","conversation_detail","create_diary_title","diary_listof_related_service",
        "doctor_all_cases","hospital_all_cases","my_favor","my_order","order_detail","personal_store_diary_list","received_votes",
        "topic_detail","welfare_detail","welfare_list","welfare_special","wiki_detail","zone_detail",
      "expert_detail","free_activity_detail","home","message_home","my_diary","organization_detail","other_homepage","question_detail",
      "search_result_diary","search_result_more","welfare_detail","zone_v3")
      for( a <- referrer ){
        val diary_clk_temp = sc.sql(
          s"""
             |select ov.partition_date,count(ov.cl_id) as clk_num,count(distinct(ov.cl_id)),count(ov.cl_id)/count(distinct(ov.cl_id))
             |from online.tl_hdfs_maidian_view ov left join final_id
             |on ov.cl_id = final_id.device_id
             |where ov.action = "page_view"
             |and params['page_name']="diary_detail"
             |and params['referrer']='${a}'
             |and ov.cl_id != "NULL"
             |and ov.partition_date >='20181201'
             |and final_id.device_id is  null
             |group by ov.partition_date
             |order by ov.partition_date
       """.stripMargin
        )
        println("来源：",a)
        diary_clk_temp.show(80)

      }



      //5.登录人数
      val log_device_temp = sc.sql(
        s"""
           |select oe.stat_date,count(distinct(oe.device_id)) as log_num
           |from data_feed_exposure oe left join final_id
           |on oe.device_id = final_id.device_id
           |and oe.stat_date >='2018-11-01'
           |and final_id.device_id is null
           |group by oe.stat_date
           |order by oe.stat_date
         """.stripMargin
      )
      println("登录人数统计：")
      log_device_temp.show(80)

    }


  }

}






object ARPU_COM {

  Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
  Logger.getLogger("org.apache.eclipse.jetty.server").setLevel(Level.OFF)

  case class Params(env: String = "dev",
                    date: String = "2018-08-01"
                   ) extends AbstractParams[Params] with Serializable

  val defaultParams = Params()

  val parser = new OptionParser[Params]("Feed_EDA") {
    head("WeafareStat")
    opt[String]("env")
      .text(s"the databases environment you used")
      .action((x, c) => c.copy(env = x))
    opt[String] ("date")
      .text(s"the date you used")
      .action((x,c) => c.copy(date = x))
    note(
      """
        |For example, the following command runs this app on a tidb dataset:
        |
        | spark-submit --class com.gmei.WeafareStat ./target/scala-2.11/feededa-assembly-0.1.jar \
      """.stripMargin +
        s"|   --env ${defaultParams.env}"
    )
  }

  def main(args: Array[String]): Unit = {
    parser.parse(args, defaultParams).map { param =>
      GmeiConfig.setup(param.env)
      val spark_env = GmeiConfig.getSparkSession()
      val sc = spark_env._2

      val ti = new TiContext(sc)
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "diary_video")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "data_feed_click")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "blacklist")
      ti.tidbMapTable(dbName = "jerry_test", tableName = "bl_device_list")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "data_feed_exposure")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "merge_queue_table")


      import sc.implicits._
      val stat_date = GmeiConfig.getMinusNDate(1)
      //println(param.date)
      val partition_date = stat_date.replace("-","")

      val agency_id = sc.sql(
        s"""
           |SELECT DISTINCT(cl_id) as device_id
           |FROM online.ml_hospital_spam_pv_day
           |WHERE partition_date >= '20180402'
           |AND partition_date <= '${partition_date}'
           |AND pv_ratio >= 0.95
           |UNION ALL
           |SELECT DISTINCT(cl_id) as device_id
           |FROM online.ml_hospital_spam_pv_month
           |WHERE partition_date >= '20171101'
           |AND partition_date <= '${partition_date}'
           |AND pv_ratio >= 0.95
         """.stripMargin
      )
      agency_id.createOrReplaceTempView("agency_id")

      val blacklist_id = sc.sql(
        s"""
           |SELECT device_id
           |from blacklist
         """.stripMargin
      )
      blacklist_id.createOrReplaceTempView("blacklist_id")

      val final_id = sc.sql(
        s"""
           |select device_id
           |from agency_id
           |UNION ALL
           |select device_id
           |from blacklist_id
         """.stripMargin
      )
      final_id.createOrReplaceTempView("final_id")


      val diary_clk_all = sc.sql(
        s"""
           |select sum(md.gengmei_price) as pay_all,count(distinct(md.device_id)) as consum_num
           |from online.ml_meigou_order_detail md left join final_id
           |on md.device_id = final_id.device_id
           |where md.status= 2
           |and final_id.device_id is  null
           |and md.partition_date = '20181218'
           |and  md.pay_time  is  not  null
           |and  md.pay_time  >=  '2018-11-01'
           |and md.pay_time  <=  '2018-11-30'
       """.stripMargin
      )
      diary_clk_all.show(80)


      val active_num = sc.sql(
        s"""
           |select count(distinct(device_id)) as active_num
           |from online.ml_device_month_active_status
           |where partition_date = '20181130'
           |and first_channel_source_type not in ('yqxiu1','yqxiu2','yqxiu3','yqxiu4','yqxiu5','mxyc1','mxyc2','mxyc3','wanpu','jinshan','jx','maimai','zhuoyi','huatian','suopingjingling','mocha','mizhe','meika','lamabang','js-az1','js-az2','js-az3','js-az4','js-az5','jfq-az1','jfq-az2','jfq-az3','jfq-az4','jfq-az5','toufang1' ,'toufang2','toufang3','toufang4','toufang5','toufang6','TF-toufang1','TF-toufang2','TF-toufang3','TF-toufang4','TF-toufang5','tf-toufang1','tf-toufang2','tf-toufang3','tf-toufang4','tf-toufang5','benzhan','promotion_aso100' ,'promotion_qianka','promotion_xiaoyu','promotion_dianru','promotion_malioaso','promotion_malioaso-shequ' ,'promotion_shike','promotion_julang_jl03')
       """.stripMargin
      )
      active_num.show(80)

    }


  }

}




object hospital_gengmei {

  Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
  Logger.getLogger("org.apache.eclipse.jetty.server").setLevel(Level.OFF)

  case class Params(env: String = "dev",
                    date: String = "2018-08-01"
                   ) extends AbstractParams[Params] with Serializable

  val defaultParams = Params()

  val parser = new OptionParser[Params]("Feed_EDA") {
    head("WeafareStat")
    opt[String]("env")
      .text(s"the databases environment you used")
      .action((x, c) => c.copy(env = x))
    opt[String] ("date")
      .text(s"the date you used")
      .action((x,c) => c.copy(date = x))
    note(
      """
        |For example, the following command runs this app on a tidb dataset:
        |
        | spark-submit --class com.gmei.WeafareStat ./target/scala-2.11/feededa-assembly-0.1.jar \
      """.stripMargin +
        s"|   --env ${defaultParams.env}"
    )
  }

  def main(args: Array[String]): Unit = {
    parser.parse(args, defaultParams).map { param =>
      GmeiConfig.setup(param.env)
      val spark_env = GmeiConfig.getSparkSession()
      val sc = spark_env._2

      val ti = new TiContext(sc)
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "diary_video")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "data_feed_click")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "blacklist")
      ti.tidbMapTable(dbName = "jerry_test", tableName = "bl_device_list")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "data_feed_exposure")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "merge_queue_table")


      import sc.implicits._
      val stat_date = GmeiConfig.getMinusNDate(1)
      //println(param.date)
      val partition_date = stat_date.replace("-","")

      val hospital_gengmei = sc.sql(
        s"""
           |SELECT id,name,location,city_id
           |FROM online.tl_hdfs_hospital_view
           |WHERE partition_date = '20181219'
         """.stripMargin
      )
      hospital_gengmei.show()
      GmeiConfig.writeToJDBCTable(hospital_gengmei, "hospital_gengmei", SaveMode.Append)

    }


  }

}






object meigou_xiaofei_renshu {

  Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
  Logger.getLogger("org.apache.eclipse.jetty.server").setLevel(Level.OFF)

  case class Params(env: String = "dev",
                    date: String = "2018-08-01"
                   ) extends AbstractParams[Params] with Serializable

  val defaultParams = Params()

  val parser = new OptionParser[Params]("Feed_EDA") {
    head("WeafareStat")
    opt[String]("env")
      .text(s"the databases environment you used")
      .action((x, c) => c.copy(env = x))
    opt[String] ("date")
      .text(s"the date you used")
      .action((x,c) => c.copy(date = x))
    note(
      """
        |For example, the following command runs this app on a tidb dataset:
        |
        | spark-submit --class com.gmei.WeafareStat ./target/scala-2.11/feededa-assembly-0.1.jar \
      """.stripMargin +
        s"|   --env ${defaultParams.env}"
    )
  }

  def main(args: Array[String]): Unit = {
    parser.parse(args, defaultParams).map { param =>
      GmeiConfig.setup(param.env)
      val spark_env = GmeiConfig.getSparkSession()
      val sc = spark_env._2

      val ti = new TiContext(sc)
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "diary_video")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "data_feed_click")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "blacklist")
      ti.tidbMapTable(dbName = "jerry_test", tableName = "bl_device_list")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "data_feed_exposure")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "merge_queue_table")



      sc.sparkContext.addJar("hdfs:///user/hive/share/lib/udf/daybits-1.0.0-SNAPSHOT.jar")
      sc.sparkContext.addJar("hdfs:///user/hive/share/lib/udf/hive-udf-1.0-SNAPSHOT.jar")


      sc.sql("CREATE TEMPORARY FUNCTION dayBitsGetW1 AS 'com.gmei.data.daybits.udf.DayBitsGetsW1'")
      sc.sql("SELECT order_id FROM mining.ml_order_spam_recognize WHERE partition_date='20181212' AND self_support=0 AND dayBitsGetW1(predict_result,'20181212')=0").show()


      import sc.implicits._
      val stat_date = GmeiConfig.getMinusNDate(1)
      //println(param.date)
      val partition_date = stat_date.replace("-","")

      val agency_id = sc.sql(
        s"""
           |SELECT DISTINCT(cl_id) as device_id
           |FROM online.ml_hospital_spam_pv_day
           |WHERE partition_date >= '20180402'
           |AND partition_date <= '${partition_date}'
           |AND pv_ratio >= 0.95
           |UNION ALL
           |SELECT DISTINCT(cl_id) as device_id
           |FROM online.ml_hospital_spam_pv_month
           |WHERE partition_date >= '20171101'
           |AND partition_date <= '${partition_date}'
           |AND pv_ratio >= 0.95
         """.stripMargin
      )
      agency_id.createOrReplaceTempView("agency_id")

      val blacklist_id = sc.sql(
        s"""
           |SELECT device_id
           |from blacklist
         """.stripMargin
      )
      blacklist_id.createOrReplaceTempView("blacklist_id")

      val final_id = sc.sql(
        s"""
           |select device_id
           |from agency_id
           |UNION ALL
           |select device_id
           |from blacklist_id
         """.stripMargin
      )
      final_id.createOrReplaceTempView("final_id")


//      val meigou_price = sc.sql(
//        s"""
//           |select  md.user_id,sum(md.gengmei_price) as pay_all
//           |from online.ml_meigou_order_detail md left join final_id
//           |on md.device_id = final_id.device_id
//           |where md.status= 2
//           |and final_id.device_id is  null
//           |and md.partition_date = '20181223'
//           |and  md.pay_time  is  not  null
//           |and  md.validate_time>'2017-01-01 00:00:00.0'
//           |group by md.user_id
//           |order by sum(md.gengmei_price)
//       """.stripMargin
//      )
//      meigou_price.show(80)


      val meigou_price = sc.sql(
        s"""
           |select  md.user_id,sum(md.gengmei_price) as pay_all
           |from online.ml_meigou_order_detail md
           |left join
           |(
           |        SELECT
           |        order_id
           |        FROM mining.ml_order_spam_recognize
           |        WHERE partition_date='20181223' AND
           |        self_support=0 AND dayBitsGetW1(predict_result,'20181223')=0
           |)spam
           |on md.order_id = spam.order_id
           |where md.status= 2
           |and  spam.order_id is null
           |and md.partition_date = '20181223'
           |and  md.pay_time  is  not  null
           |and  md.validate_time>'2017-01-01 00:00:00.0'
           |group by md.user_id
           |order by sum(md.gengmei_price)
       """.stripMargin
      )
//      meigou_price.show(80)

//      GmeiConfig.writeToJDBCTable(meigou_price, "meigou_price", SaveMode.Overwrite)

    }


  }

}




object smart_rank_count {

  Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
  Logger.getLogger("org.apache.eclipse.jetty.server").setLevel(Level.OFF)

  case class Params(env: String = "dev",
                    date: String = "2018-08-01"
                   ) extends AbstractParams[Params] with Serializable

  val defaultParams = Params()

  val parser = new OptionParser[Params]("Feed_EDA") {
    head("WeafareStat")
    opt[String]("env")
      .text(s"the databases environment you used")
      .action((x, c) => c.copy(env = x))
    opt[String] ("date")
      .text(s"the date you used")
      .action((x,c) => c.copy(date = x))
    note(
      """
        |For example, the following command runs this app on a tidb dataset:
        |
        | spark-submit --class com.gmei.WeafareStat ./target/scala-2.11/feededa-assembly-0.1.jar \
      """.stripMargin +
        s"|   --env ${defaultParams.env}"
    )
  }

  def main(args: Array[String]): Unit = {
    parser.parse(args, defaultParams).map { param =>
      GmeiConfig.setup(param.env)
      val spark_env = GmeiConfig.getSparkSession()
      val sc = spark_env._2

      val ti = new TiContext(sc)
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "diary_video")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "data_feed_click")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "blacklist")
      ti.tidbMapTable(dbName = "jerry_test", tableName = "bl_device_list")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "data_feed_exposure")
      ti.tidbMapTable(dbName = "jerry_prod", tableName = "merge_queue_table")


      import sc.implicits._
      val stat_date = GmeiConfig.getMinusNDate(1)
      //println(param.date)
      val partition_date = stat_date.replace("-","")

      val agency_id = sc.sql(
        s"""
           |SELECT DISTINCT(cl_id) as device_id
           |FROM online.ml_hospital_spam_pv_day
           |WHERE partition_date >= '20180402'
           |AND partition_date <= '${partition_date}'
           |AND pv_ratio >= 0.95
           |UNION ALL
           |SELECT DISTINCT(cl_id) as device_id
           |FROM online.ml_hospital_spam_pv_month
           |WHERE partition_date >= '20171101'
           |AND partition_date <= '${partition_date}'
           |AND pv_ratio >= 0.95
         """.stripMargin
      )
      agency_id.createOrReplaceTempView("agency_id")

      val blacklist_id = sc.sql(
        s"""
           |SELECT device_id
           |from blacklist
         """.stripMargin
      )
      blacklist_id.createOrReplaceTempView("blacklist_id")

      val final_id = sc.sql(
        s"""
           |select device_id
           |from agency_id
           |UNION ALL
           |select device_id
           |from blacklist_id
         """.stripMargin
      )
      final_id.createOrReplaceTempView("final_id")


      val meigou_pv = sc.sql(
        s"""
           |select device,params
           |from online.tl_hdfs_maidian_view
           |where action = "page_view"
           |and params['page_name']="welfare_detail"
           |and partition_date ='20181201'
           |LIMIT 3
       """.stripMargin
      )
       meigou_pv.show(80)

//      val zixun_pv = sc.sql(
//        s"""
//           |select cl_id,city_id,params['service_id'] as meigou_id
//           |from online.tl_hdfs_maidian_view
//           |where partition_date > '20171118'
//           |and action = 'welfare_detail_click_message'
//       """.stripMargin
//      )
//      zixun_pv.show(80)



//      val meigou_price = sc.sql(
//        s"""
//           |select  md.user_id,sum(md.gengmei_price) as pay_all
//           |from online.ml_meigou_order_detail md
//           |left join
//           |(
//           |        SELECT
//           |        order_id
//           |        FROM mining.ml_order_spam_recognize
//           |        WHERE partition_date='20181223' AND
//           |        self_support=0 AND dayBitsGetW1(predict_result,'20181223')=0
//           |)spam
//           |on md.order_id = spam.order_id
//           |where md.status= 2
//           |and  spam.order_id is null
//           |and md.partition_date = '20181223'
//           |and  md.pay_time  is  not  null
//           |and  md.validate_time>'2017-01-01 00:00:00.0'
//           |group by md.user_id
//           |order by sum(md.gengmei_price)
//       """.stripMargin
//      )

      //      GmeiConfig.writeToJDBCTable(meigou_price, "meigou_price", SaveMode.Overwrite)

    }


  }

}