package com.aliyun.odps.spark.sparksql

import config.MyConfigSession


import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.apache.spark.storage.StorageLevel
import utils.{DateUtils, StringUtils, UseUtil}

import scala.collection.mutable
import scala.collection.mutable.ListBuffer
import scala.util.control.Breaks.{break, breakable}


/**
 * 处理昨天的数据,用于计算时长以及访问路径，结果导入： dw_fact_log_session_term
 * 注意点：
 * 1.只保留service_name='trace2'(native)埋点数据，（h5埋点数据也会走native上报一份）
 * 2.去掉ACTION_EQUIP_INFO类数据
 * 1.首页加载前的引导页(android不埋 )不计入首页流量统计
 * 2.对于ACTION_VIEW类型事件,用页面的resume以及stop事件计算访问时差
 * 3.对于ACTION_HEART_BEAT事件中的back类型或者是BACKGROUND类事件，将menu_code定位back，
 * 取连续backgroud的末次back的时间以及首次back时间差作为后台执行时长
 * 4.如果menu_code值为空或值不符合规范，如果view_class有值取view_class代替
 *
 * @Author weicheng.mao
 * @Date 2022/03/02 10:58
 * @Version 1.0
 */

object SessionProcessTerm {
  def apply(): SessionProcessTerm = new SessionProcessTerm()

  def main(args: Array[String]): Unit = {

    //设置同步数据的批次号,格式是2019-09-12
    var scnData: String = DateUtils.getYesterdayDate
    var elseFiler = " 1=1"
    if (args.length >= 1) {
      scnData = args(0)
      if(args.length > 1 && args(1)!=""){
        elseFiler = args(1)
      }
    }
    println(s"scnData=${scnData}")
    //设置任务开始时间,格式是2019-09-12 14:03:30
    val startTime: String = DateUtils.getTodayTime
    //存储SQL中的参数
    val insertArr: Array[String] = Array[String](scnData, startTime)

    val sessionProcessTerm: SessionProcessTerm = SessionProcessTerm()
     try {
      val sparkSession: SparkSession = sessionProcessTerm.getSparkSession("SessionProcessTerm")

       //获取menu_code广播变量
       val menuCodeBroad = UseUtil.getBroadcast(sparkSession, MyConfigSession.MENU_CODE_SQL, "view_path", "menu_code")
       //    //获取actionCategory变量
       val actionCategory = UseUtil.getBroadcast(sparkSession, MyConfigSession.ACTION_CATEGORY_SQL, "action_type", "action_category")
       //获取position对应的label_value广播变量
       val positionUrlLabelBroad = UseUtil.getBroadcast(sparkSession, MyConfigSession.ACTION_URLLABEL_SQL, "url_content", "label_value")
       println(s"positionUrlLabelBroad=${positionUrlLabelBroad.value}")
      var SOURCE_SQL_COMMON =
        s"""
           |select  pseudo_session,doctor_id,mobile,device_token,class_name,view_path,action,
           |component_tag,app_version,device_type, cast(created as bigint) created,user_token_tourist,
           |network_type ,device_brand,device_model,alternate_info,remark2 first_app_version,serviceName service_name
           |from pica.picalog_trace_app_part
           | where  pseudo_session is not null and pseudo_session !='' and pseudo_id !='' and extra_info !='com.picahealth.patient'
           | and action!='ACTION_EQUIP_INFO'
           | and action!='ACTION_WEB_AFTER'
           |  and created is not null and created!='' and FROM_UNIXTIME(cast(substring(created,1,10) as bigint),'yyyy-MM-dd')='${scnData}'
           |   and created_day='${scnData}' and ${elseFiler}
           |""".stripMargin
      var SOURCE_SQL_TERM = SOURCE_SQL_COMMON + " and  serviceName='trace2' "
       println("SOURCE_SQL_TERM=="+SOURCE_SQL_TERM )
       var sourceDF: DataFrame = sparkSession.sql(SOURCE_SQL_TERM )
       /*
        补充站外埋点数据
        */
       //part1.通过device_token补充,device_token不为空时对应的埋点只有trace1数据的需要加进来
      val SOURCE_SQL_TERM_ADD =
        s"""
          |select  b.*  from (
            | select device_token,   concat_ws( '_',collect_set(service_name) )  cws
            | from  (${SOURCE_SQL_COMMON}) s group by device_token  having concat_ws( '_',collect_set(service_name)) ='trace1'
            |  ) a
          | join ( ${SOURCE_SQL_COMMON + " and  serviceName='trace1' "}) b on a.device_token=b.device_token
          |""".stripMargin
       println("SOURCE_SQL_TERM_ADD=="+SOURCE_SQL_TERM_ADD)
       var sourceAddDF =  sparkSession.sql(SOURCE_SQL_TERM_ADD )
       sourceAddDF.createOrReplaceTempView("source_add_table")
       //part2.通过pseudo_session补充,device_token为空时,pseudo_session对应的埋点只有trace1数据且doctor_id不为空的需要加进来
       val SOURCE_SQL_TERM_ADD2 =
         s"""
            |select  b.* from ( ${SOURCE_SQL_COMMON + " and  serviceName='trace1' "}) b where b.doctor_id in (
            | select doctor_id
            | from  (${SOURCE_SQL_COMMON +" and doctor_id not in ('','0') and doctor_id is not null" }) s group by doctor_id  having concat_ws( '_',collect_set(service_name)) ='trace1'
            |  )
            | and   b.pseudo_session not in (select  pseudo_session from source_add_table group by pseudo_session )
            |""".stripMargin
       println("SOURCE_SQL_TERM_ADD2=="+SOURCE_SQL_TERM_ADD2)
       var sourceAdd2DF =  sparkSession.sql(SOURCE_SQL_TERM_ADD2 )
      val sourceResDF = sourceDF.union(sourceAddDF).union(sourceAdd2DF)
      println("sourceResDF.show==================")
       sourceResDF.printSchema()
      println("selectDF.count=========",sourceResDF.count())
      var conditionGroup = List( "<='1'","between '2' and '3'","between '4' and '5'","between '6' and '7'","between '8' and '9'",
          "between 'a' and 'b'","between 'c' and 'd'",">='e'" )
      var dataCount = 0
      var index = 0
       sourceResDF.persist(StorageLevel.MEMORY_AND_DISK_SER)

      for(condition <- conditionGroup){
        index += 1
        val slideDF = sourceResDF.where(s" SUBSTRING(pseudo_session,2,1)  ${condition}").repartition(100)
        println(s"-----------------------------------compute refer columns,condition=${condition}-----------------------------------------")
        println(s"-----------------------------------compute refer columns,condition_count=${slideDF.count()}-----------------------------------------")

        val baseDF = sessionProcessTerm.offsetValues(menuCodeBroad,actionCategory,positionUrlLabelBroad,slideDF)
        baseDF.repartition(120).persist(StorageLevel.MEMORY_AND_DISK_SER)
        val referResultRdd = sessionProcessTerm.getReferColumns(baseDF)
        val referResultDF: DataFrame = sparkSession.createDataFrame(referResultRdd, StructType(
          List(StructField("pseudo_session", StringType, false),
            StructField("session_id", StringType, false),
            StructField("device_token", StringType, false),
            StructField("user_id", IntegerType, false),
            StructField("user_token", StringType, false),
            StructField("mobile", StringType, false),
            StructField("menu_code", StringType, false),
            StructField("menu_begin_time", StringType, false),
            StructField("action_code", StringType, false),
            StructField("position", StringType, false),
            StructField("label_value", StringType, false),
            StructField("action", StringType, false),
            StructField("action_type", StringType, false),
            StructField("action_step", StringType, false),
            StructField("device_type", StringType, false),
            StructField("app_version", StringType, false),
            StructField("created_time", LongType, false),
            StructField("date_time", StringType, false),
            StructField("pre_session_id", StringType, false),
            //新增字段
            StructField("label_class", StringType, true),
            StructField("net_type", StringType, true),
            StructField("module_class1", StringType, true),
            StructField("module_class2", StringType, true),
            StructField("device_brand", StringType, true),
            StructField("device_model", StringType, true),
            StructField("view_class", StringType, true),
            StructField("view_path", StringType, true),
            StructField("alternate_info", StringType, true),
            StructField("first_app_version", StringType, true),
            StructField("service_name", StringType, true),
            StructField("tag8", StringType, true),
            StructField("tag9", StringType, true),
            StructField("component_tag", StringType, true)
          )
        ))
        println("referResultDF.show()============'")
        referResultDF.show()
        referResultDF.printSchema()
        referResultDF.persist(StorageLevel.MEMORY_AND_DISK_SER).createOrReplaceTempView("refer_result_table")
        println("-----------------------------------compute menu_code term-----------------------------------------")
        val getMenuTermSql =
          """
            |select  a.session_id,a.device_token,a.user_id,a.menu_code,
            |lag(a.created_time) over(partition by a.session_id order by a.created_time desc ) menu_end_time,a.created_time
            | from  refer_result_table a
            | left  join(select session_id,min(app_version) min_version from refer_result_table group by session_id) b on a.session_id=b.session_id
            |""".stripMargin
        //处理session最小版本>='3.4.5'的session数据
        val newVersionMenuDF = sparkSession.sql(s"${getMenuTermSql} where b.min_version>='3.4.5' and a.action in('ACTION_ACTIVITY_RESUME','ACTION_HEART_BEAT','ACTION_WEB_ENTER') ")
        println("newVersionMenuDF,show()======")
        val oldVersionMenuDF = sparkSession.sql(s"${getMenuTermSql} where b.min_version<'3.4.5' and a.action_type in ('ACTION_VIEW','ACTION_HEART')")
        println("oldVersionMenuDF,show()======")
        val referMenuDF = newVersionMenuDF.union(oldVersionMenuDF)
        println("referMenuDF.show()=========")
        referMenuDF.printSchema()
        referMenuDF.createOrReplaceTempView("refer_menu_table")

        println("-----------------------------------compute session_id end_time-----------------------------------------")
        val sessionEndDF: DataFrame = sessionProcessTerm.getSessionTail(slideDF, sparkSession)
        println("------sessionEndDF.show()-------------")
        sessionEndDF.printSchema()
        sessionEndDF.createOrReplaceTempView("session_end_table")
        val getReferSql =
          s"""
             |select
             |a.session_id,
             |a.device_token,
             |a.user_id,
             |a.user_token,
             |a.mobile,
             |a.menu_code,
             |a.action_code,
             |a.position,
             |a.label_value,
             |a.label_class,
             |a.module_class1,
             |a.module_class2,
             |a.action_type,
             |a.menu_begin_time,
             | case when  a.action_type in('ACTION_VIEW','ACTION_HEART') and  b.menu_end_time is null  then c.session_end_time else b.menu_end_time end menu_end_time,
             |(cast((case when  a.action_type in('ACTION_VIEW','ACTION_HEART') and  b.menu_end_time is null  then c.session_end_time else b.menu_end_time end) as bigint)-cast(a.menu_begin_time as bigint))/1000 menu_time_diff,
             |a.action_step,
             |a.device_type,
             |a.device_brand,
             |a.device_model,
             |a.app_version,
             |a.net_type,
             |a.created_time,
             |a.date_time,
             |c.session_begin_time,
             |c.session_end_time,
             |(cast(c.session_end_time as bigint)-cast(c.session_begin_time as bigint))/1000 session_time_diff,
             |a.pre_session_id,
             |case when a.user_id=0 then a.device_token else a.user_id end user_identity_id,
             |a.view_class,a.view_path, a.alternate_info, a.first_app_version,a.service_name,a.tag8,a.tag9,a.component_tag
             |from refer_result_table a
             |left join refer_menu_table b on a.session_id=b.session_id and a.device_token=b.device_token and a.user_id=b.user_id and a.menu_code=b.menu_code  and a.created_time=b.created_time and a.action_type in('ACTION_VIEW','ACTION_HEART')
             |left join session_end_table c on a.session_id = c.session_id
             | distribute by rand()
             |""".stripMargin
        val menuCodeAddDF = sparkSession.sql(getReferSql).repartition(100)
        println("menuCodeAddDF.show=======")
        menuCodeAddDF.printSchema()
        menuCodeAddDF.persist(StorageLevel.MEMORY_AND_DISK_SER)
        println("-------------------------------match user_id 逻辑-------------------------------------------------")
        val resultDF: DataFrame = sessionProcessTerm.matchUserId(menuCodeAddDF, sparkSession, scnData).repartition(5)
        println("dwFactLogSession.show=======>")
        resultDF.printSchema()
        println("-----------------------------------load data to pica_dw.dw_fact_log_session_term-----------------")
        sessionProcessTerm.loadData(resultDF, sparkSession, scnData,index,dataCount)
        val resCount = resultDF.count().toInt
        println(s"${condition}的结果==${resCount}")
        dataCount += resCount
      }
      println("----------------------------------update task record table---------------------------------------")
      //任务执行成功,更新 Mysql record 配置表
      val updateSQL: String =
        s"""
           |update ${MyConfigSession.JDBC_TABLE} set status=?,end_time=?,data_count=? where job_name='${MyConfigSession.HIVE_TABLE4}' and start_time='${startTime}'
				""".stripMargin

      sparkSession.stop()
    } catch {
      case e: Exception => {
        println("-----------------------------------任务异常---------------------------------------------------")
        e.printStackTrace()
        val exceptionSQL: String =
          s"""
             |update ${MyConfigSession.JDBC_TABLE} set status=?,exception=?,end_time=? where job_name='${MyConfigSession.HIVE_TABLE4}' and start_time='${startTime}'
					""".stripMargin

      }
    }
  }
}

class SessionProcessTerm {

  def getSparkSession(appName: String): SparkSession = {
//    val conf: SparkConf = new SparkConf().setAppName(appName)
//    UseUtil.setConfigure(conf)
//    val sparkSession: SparkSession = SparkSession.builder().config(conf).enableHiveSupport().getOrCreate()
//    sparkSession

  val sparkSession = SparkSession
    .builder()
    .config("spark.hadoop.odps.project.name", "pica")
    .appName("SessionProcessTerm")
    .getOrCreate()
    sparkSession
  }

  def offsetValues(menuCodeBroad:Broadcast[Map[String,String]], actionCategory:Broadcast[Map[String,String]],
                   positionUrlLabelBroad:Broadcast[Map[String,String]],dataFrame: DataFrame): DataFrame = {
    val path_menu: Map[String, String] = menuCodeBroad.value //关联到menu_code的映射表广播变量
    val actionCategoryMap: Map[String, String] = actionCategory.value //关联到action_category的映射表广播变量
    val positionLabelMap: Map[String, String] = positionUrlLabelBroad.value

    val groupRdd = dataFrame.rdd.groupBy(r => r.getAs[String]("pseudo_session"))

    val baseRdd = groupRdd.flatMap(g => {
      val pseudo_session = g._1
      val resList: ListBuffer[SessionTerm] = new ListBuffer[SessionTerm]()
      var rowList = g._2
      rowList = rowList.toList.sortWith((x, y) => x.getAs[Long]("created") > y.getAs[Long]("created")) //按created由大到小排序
      var thisDeviceToken = ""
      var thisDoctorId = "0"
      var thisMobile = ""
      rowList.foreach(row => {
        var deviceToken = row.getAs[String]("device_token")
        var doctorId = row.getAs[String]("doctor_id")
        var mobile = row.getAs[String]("mobile")
        if (deviceToken != null && !deviceToken.equals("")) {
          thisDeviceToken = deviceToken
        } else {
          deviceToken = thisDeviceToken
        }
         if (doctorId != null && !doctorId.equals("") && !doctorId.equals("0")) {
          thisDoctorId = doctorId
        } else {
          doctorId = thisDoctorId
        }
        if (mobile != null && !mobile.equals("")) {
          thisMobile = mobile
        } else {
          mobile = thisMobile
        }
        //1.获取网络类型
        //2G,3G,4G,2G/3G/4G,WIFI,WLAN,或者为空字符串
        val net_type = UseUtil.netTypeMatch(StringUtils.getNotNullString(row.getAs[String]("network_type")))
        //2.修改action类型,保留原始字段
        val action = row.getAs[String]("action")
        var action_type: String = ""
        if (action != null) {
          action_type = actionCategoryMap.getOrElse(action, "ACTION_UNKNOWN")
          if(action=="ACTION_HEART_BEAT"){
            action_type="ACTION_HEART"
          }
        }
        //3.拆分 component_tag字段
        val component_tag: String = StringUtils.getNotNullString(row.getAs[String]("component_tag"))
        val view_path = StringUtils.getNotNullString(row.getAs[String]("view_path"))
        val tagArr = Array("menu_code", "action_code", "position", "label_value", "label_class", "module_class1", "module_class2", "tag8", "tag9")
        val tagMap = mutable.Map[String, String]()
        tagArr.foreach(r => tagMap.put(r, ""))
        //针对特定类型事件,将符合要求的component_tag进行切割,获取 aciton_code,label_value
        if(List("ACTION_VIEW", "ACTION_CLICK", "ACTION_EXPOSE").contains(action_type)){
          if (component_tag.contains("#")) {//默认是
            //按照#号切割
            val strs: Array[String] = component_tag.split("#")
            var index = 0
            for (value <- strs) {
              val filedName = tagArr.apply(index)
              if ("label_value".equals(filedName)) {
                tagMap.put(filedName, value.substring(0, math.min(250, strs(3).length)))
              } else {
                tagMap.put(filedName, value)
              }
              index += 1
            }
          }else if(component_tag.length<5  && StringUtils.isNumeric(component_tag) && component_tag.toInt>0  ){ //如果component_tag不为0,且长度在合理区间,则作为menu_code值
            tagMap.put("menu_code",component_tag)
          }
        }
        var menu_code_new = tagMap("menu_code")
        if ("MenuCode_081".equals(menu_code_new)) {
          menu_code_new = "081" //针对异常menu_code值单独处理
        }
        //匹配menu_code:如果上述截取出来的menu_code为(''||null||0||length(menu_code)>3  && !component_tag.contains("#") ) and action is ACTION_VIEW
        if ((menu_code_new.equals("") || menu_code_new.equals("null") || menu_code_new.equals("0") || (menu_code_new.length > 3  && !component_tag.contains("#")) )
          && action_type.equals("ACTION_VIEW")) {
          menu_code_new = "0" //关联不上的显示为0
          import scala.util.control.Breaks._
          breakable {
            //利用menu_code映射表匹配
            for (tuple <- path_menu) {
              //源数据view_path的字符串包含映射表view_path的字符串
              if (view_path.contains(tuple._1)) {
                //满足条件后,修改源数据的menu_code
                menu_code_new = tuple._2
                println("--------------------menu_code match successfully-----------------------")
                //结束遍历
                break()
              }
            }
          }
        }
        //解析(5分钟|健康知识)详情页的id到position
        if(action_type.equals("ACTION_VIEW")  && List("425","015").contains(menu_code_new) && view_path.contains("?")){
          val kvs=view_path.split("\\?").apply(1).split("&")
          for(kv <- kvs){
            val k = kv.split("=").apply(0)
            if(menu_code_new=="425" && k=="id"){ //五分钟详情页
              tagMap("position")= kv.split("=").apply(1)
            }
            if(menu_code_new=="015" && k=="eduComId"){ //健康知识详情页
              tagMap("position")= kv.split("=").apply(1)
            }
          }
        }
        //
        if(List("930000","930001","930002" ).contains(tagMap("action_code"))
            && "ACTION_CLICK".equals(action_type) && "930"== menu_code_new){
          println("------------------------------------单独计算label_value----------------------------------------------")
          //"menu_code = '930' and  action_code IN ( '930000', '930001', '930002' ) and action_type = 'ACTION_CLICK'
          breakable {
            //利用position url_content映射表匹配
            for (tuple <- positionLabelMap) {
              if (StringUtils.getNotNullString(tagMap("position")).contains(tuple._1) && tuple._2!="" && tuple._2 != null) {
                //满足条件后,修改源数据的label_value
                tagMap("label_value") = tuple._2
                println("--------------------menu_code match successfully-----------------------")
                //结束遍历
                break()
              }
            }
          }
        }
        //针对menu_code缺失情况补偿处理
        var menu_code_offset = menu_code_new
        if(component_tag=="back"){
          menu_code_offset = "back"
        }
        val view_class = StringUtils.getNotNullString(row.getAs[String]("class_name"))
        if( List("0","null","").contains(menu_code_new) &&  !List("0","null","").contains(view_class)){
          menu_code_offset = view_class
        }
        var user_id  = 0
        if(doctorId != null && !doctorId.equals("") && !doctorId.equals("0")){
          user_id = doctorId.toInt
        }
        var app_version = StringUtils.getNotNullString(row.getAs[String]("app_version"))
        var is_need = true //添加一个字段,下一步计算时长使用,
        // 针对3.4.5之后的版本,使用ACTION_ACTIVITY_RESUME计算页面起止
        if(app_version.length<8 && app_version>="3.4.5" && action=="ACTION_ACTIVITY_CREATE" && menu_code_new!="017" ){
          is_need =  false
        }
        resList += SessionTerm(pseudo_session,
          user_id,
          mobile,
          deviceToken,
          StringUtils.getNotNullString(row.getAs[String]("user_token_tourist")),
          view_class,
          StringUtils.getNotNullString(row.getAs[String]("view_path")),
          action, action_type, component_tag, tagMap("menu_code"), menu_code_new,menu_code_offset, tagMap("action_code"),
          tagMap("position"), tagMap("label_value"), tagMap("label_class"), tagMap("module_class1"), tagMap("module_class2"),
          app_version,
          StringUtils.getNotNullString(row.getAs[String]("device_type")),
          StringUtils.getNotNullString(row.getAs[String]("device_brand")),
          StringUtils.getNotNullString(row.getAs[String]("device_model")),
          net_type, row.getAs[Long]("created"),
          DateUtils.milliSecondsFormatTime(row.getAs[Long]("created") + ""),
          StringUtils.getNotNullString(row.getAs[String]("alternate_info")),
          StringUtils.getNotNullString(row.getAs[String]("first_app_version")),
          StringUtils.getNotNullString(row.getAs[String]("service_name")),
          tagMap("tag8"), tagMap("tag9"), is_need
        )
      })
      resList.iterator
    })
    import dataFrame.sparkSession.implicits._
    var baseDF = baseRdd.toDF("pseudo_session", "user_id", "mobile", "device_token", "user_token", "view_class", "view_path", "action", "action_type", "component_tag",
      "menu_code", "menu_code_new","menu_code_offset", "action_code", "position", "label_value", "label_class", "module_class1", "module_class2", "app_version", "device_type", "device_brand", "device_model",
      "net_type", "created_time", "date_time", "alternate_info",  "first_app_version", "service_name", "tag8", "tag9",  "is_need")
    println("baseDF.show=======>")
    baseDF.where("menu_code_new ='017'").show()
    baseDF.printSchema()
    baseDF
  }
  /**
   * @Description 匹配user_id,补全数据中的user_id字段
   * @param dataFrame       筛选后的数据
   * @param sparkSQLSession SparkSession 环境
   * @param created_day     当前数据的日期,格式 "2020-03-01"
   * @return org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>
   **/
  def matchUserId(dataFrame: DataFrame, sparkSQLSession: SparkSession, created_day: String): DataFrame = {
    //追加:将dataFrame与pica_ds.pica_doctor根据user_id进行匹配,匹配不上的user_id置为'0'
    println("matchUserId开始执行-----------------------------------")
    dataFrame.withColumnRenamed("user_id", "user_id_ods").createOrReplaceTempView(MyConfigSession.VIEW_SESSION_ODS)
    val INIT_USER_ID_SQL_PREF =
      s"""
         |SELECT t.*,t.user_id_ods user_id_old, COALESCE(cast(b.id as string),'0') AS user_id  from ${MyConfigSession.VIEW_SESSION_ODS} as t
         |left join  (select  id,cast(id as string) id_str  from pica.ods_pica_p_doctor a where  to_date(a.creat_time) <= '${created_day}') AS b on t.user_id_ods = cast(b.id as string)
            """.stripMargin
    val DF = sparkSQLSession.sql(INIT_USER_ID_SQL_PREF).drop("user_id_ods")
    /*
        以下的所有逻辑是为了补全user_id字段
     */
    //第一步：首先筛选出不符合的use_id数据,将这些user_id置为字符串'0'
    val noMatchUserIdDF: Dataset[Row] = DF.where("user_id ='' OR user_id = '0' OR LENGTH(user_id) = 24")
    noMatchUserIdDF.drop("user_id_old").withColumnRenamed("user_id", "user_id_old")
      .createOrReplaceTempView(MyConfigSession.VIEW_SESSION_NO_MATCH)
    //1.筛选出上一步没有匹配到的user_id,先按照mobile_phone进行匹配
    val MOBILE_PHONE_SQL_PREF: String =
      s"""
         |SELECT ss.*,COALESCE(cast(b.id as string),'0') AS user_id  from ${MyConfigSession.VIEW_SESSION_NO_MATCH}  AS ss
         |left join (select distinct id,mobile_phone from pica.ods_pica_p_doctor where  delete_flag=1 and mobile_phone!='' and mobile_phone!='XK0HdMN6dAfOlYPOFHHL0A==') AS b on ss.mobile = b.mobile_phone
         """.stripMargin
    val mobilePhoneDF: DataFrame = sparkSQLSession.sql(MOBILE_PHONE_SQL_PREF)
    //2.使用临时表equiment,筛选出为1的那条最新数据
    var equipmentInfoSql = MyConfigSession.EQUIPMENT_INFO_SQL
    if (!created_day.equals(DateUtils.getYesterdayDate)) { //如果不是跑昨天的数据,使用equipment拉链表
      equipmentInfoSql = MyConfigSession.EQUIPMENT_INFO_SQL_ARGS + s"'${created_day}'"
    }
    println(s"equipmentInfoSql=${equipmentInfoSql}")
    val equipmentDF: DataFrame = sparkSQLSession.sql(equipmentInfoSql).where("row_d =1")
    equipmentDF.createOrReplaceTempView(MyConfigSession.VIEW_EQUIPMENT_INFO)
    mobilePhoneDF.drop("user_id_old").withColumnRenamed("user_id", "user_id_old")
      .createOrReplaceTempView(MyConfigSession.VIEW_MOBILE_PHONE)
    val DEVICE_TOKEN_SQL_PREF: String =
      s""" SELECT t.*,COALESCE(cast(b.user_id as string),'0') AS user_id
         | from (select * from  ${MyConfigSession.VIEW_MOBILE_PHONE}  a where a.user_id_old= '0' ) as t
         |left join ${MyConfigSession.VIEW_EQUIPMENT_INFO} as b on t.device_token = b.device_token
            """.stripMargin
    println(s"DEVICE_TOKEN_SQL_PREF=${DEVICE_TOKEN_SQL_PREF}")
    //3.将第2步筛选出来的数据按照device_token进行匹配,获得user_id
    val deviceTokenDF: DataFrame = sparkSQLSession.sql(DEVICE_TOKEN_SQL_PREF)

    //4.将上述三者union,最终导入表中的数据
    val rightUserIdDF: Dataset[Row] = DF.where("user_id !='' and user_id != '0' and LENGTH(user_id) !=24")
    val mobilePhoneResDF: Dataset[Row] = mobilePhoneDF.where("user_id !='0'")
    println("rightUserIdDF/mobilePhoneResDF/deviceTokenDF.schema===========")
    rightUserIdDF.printSchema()
    mobilePhoneResDF.printSchema()
    deviceTokenDF.printSchema()
    val dwFactLogSession: Dataset[Row] = rightUserIdDF.union(mobilePhoneResDF).union(deviceTokenDF)
    dwFactLogSession.createOrReplaceTempView(MyConfigSession.VIEW_DEVICE_TOKEN)
    dwFactLogSession
  }

  /**
   * @Description 获取需要的字段的refer字段以及对pseudo_session进行拆分
   *              https://www.tapd.cn/64812329/prong/stories/view/1164812329001012031需求上线版本为(Android:3.4.5,iOS:3.4.4,之后的版本统一到3.4.6)
   *              计算sessionId,按版本号区分数据,之前的作为老数据处理,之后的作为新数据处理,老数据对pseudo_session进行拆分,时间最大间隔30分钟
   *              新数据不拆分session,按ACTION_HEART_BEAT区分session活跃时间(新增ACTION_HEART_BEAT事件的版本是3.3.2)
   * @param sourceDF 源数据
   * @return org.apache.spark.sql.Dataset<org.apache.spark.sql.Row>
   **/
  def getReferColumns(sourceDF: DataFrame): RDD[Row] = {
    val actions = List("'ACTION_VIEW'", "'ACTION_CLICK'", "'ACTION_EXPOSE'","'ACTION_START'") //高版本使用"'ACTION_HEART_BEAT'"
    val actionCommonDF = sourceDF.where(s" ( action_type in (${actions.mkString(",")}) or (action_type='ACTION_HEART' and menu_code_offset='back' )) ")
        .filter("is_need")

    println("actionCommonDF.show()===========")
    actionCommonDF.filter("menu_code='017'").show()
    actionCommonDF.printSchema()
    //计算session中menu_code的起始时间,对于ACTION_HEART_BEAT的埋点记录剔除冗余数据
    val groupRdd = actionCommonDF.rdd.groupBy(r => r.getAs[String]("pseudo_session"))//.repartition(100)
    println("开始执行groupRdd.flatMap=========")
    val resRdd = groupRdd.flatMap(g => {
      val pseudo_session = g._1
      val resList: ListBuffer[Row] = new ListBuffer[Row]()
      var rowList = g._2
      rowList = rowList.toList.sortWith((x, y) => x.getAs[Long]("created_time") < y.getAs[Long]("created_time")) //按created由小到小排序
      var actionStep = "0_0"
      var thisMenuCode = ""
      var menuBeginTime = ""
      var prefCreatedTime = ""
      var prefActionType = ""
      var count = 0

      for (row <- rowList) {
        var isSessionNew = false
        var appVersion = row.getAs[String]("app_version")
        //如果版本号低于3.3.2,则没有ACTION_HEART_BEAT事件,需要根据两次事件之间的时差对session进行切分,超过30分钟则拆分出一个session
        var createdTime = row.getAs[Long]("created_time")
        if (appVersion < "3.3.2") { //版本号低于3.3.2,需要根据时差判断是否 拆分session
          if (prefCreatedTime != "" && (createdTime.toLong - prefCreatedTime.toLong) > MyConfigSession.SESSION_GAP) { //本次action事件事件比
            count += 1
            isSessionNew = true
          }
        }
        var menuCode = row.getAs[String]("menu_code_offset")
        val action = row.getAs[String]("action")
        val actionType = row.getAs[String]("action_type")
        var sessionId = pseudo_session + count
        var preSessionId = ""
        if (count > 0) {
          preSessionId = pseudo_session + (count - 1)
        }
        //使用版本以及action_type限制是否入库
        var needPut = true
        if ("ACTION_VIEW".equals(actionType) || "ACTION_HEART".equals(actionType)) {
          if (appVersion >= "3.4.5") { //针对3.4.5之后的版本单独处理
            //如果本条记录为ACTION_ACTIVITY_RESUME类型或者ACTION_HEART类型或者menu_code为017表示外部方式启动app,则入库
            if ("ACTION_ACTIVITY_RESUME".equals(action) || "ACTION_HEART".equals(actionType) ||"ACTION_WEB_ENTER".equals(action)|| "017".equals(menuCode)) {//有些h5页面没有对应的ACTION_ACTIVITY_RESUME,需要补进来
              needPut = true
            }else{
              needPut = false
            }
            if ( thisMenuCode.equals(menuCode) && ("ACTION_HEART"==prefActionType  || "ACTION_VIEW" == prefActionType)){
              needPut = false
            }
          }else{
            if ( thisMenuCode.equals(menuCode) && ("ACTION_HEART"==prefActionType  || "ACTION_VIEW" == prefActionType) && !isSessionNew ) {
              needPut = false
            }
          }
          if(needPut){
            menuBeginTime = createdTime.toString
          }
        }
        //如果menuCode发生改变,或者session改变,重置本轮menuCode
        if (!thisMenuCode.equals(menuCode) || isSessionNew) {
          menuBeginTime = createdTime.toString
          if(needPut ){//只有在确定该menu_code对应的记录入库后,才更新thisMenuCode以及对应的menu_code访问顺序
            thisMenuCode = menuCode
            actionStep = (actionStep.split("_")(0).toInt + 1).toString + "_0"
          }
        } else {//如果menuCode与session都不变, actionType改变或者actionType为非浏览行为,则调整actionStep
          if (!(prefActionType.equals(actionType) && "ACTION_VIEW".equals(actionType))) {
            actionStep = (actionStep.split("_")(0)) + "_" + (actionStep.split("_")(1).toInt + 1).toString
          }
        }
//        println(s"created_time=${createdTime},thisMenuCode=${thisMenuCode},actionType=${actionType},action=${action},isMenuChange=${!thisMenuCode.equals(menuCode)},prefActionType=${prefActionType},needPut=${needPut}" )
        prefCreatedTime = createdTime.toString
        prefActionType = actionType
        if (needPut) {
          resList += (Row( pseudo_session,sessionId,
            row.getAs[String]("device_token"),
            row.getAs[Integer]("user_id"),
            row.getAs[String]("user_token"),
            row.getAs[String]("mobile"),
            thisMenuCode,
            menuBeginTime,
            row.getAs[String]("action_code"),
            row.getAs[String]("position"),
            row.getAs[String]("label_value"),
            action, actionType, actionStep,
            row.getAs[String]("device_type"),
            row.getAs[String]("app_version"),
            createdTime,
            row.getAs[String]("date_time"),
            preSessionId,
            row.getAs[String]("label_class"),
            row.getAs[String]("net_type"),
            row.getAs[String]("module_class1"),
            row.getAs[String]("module_class2"),
            row.getAs[String]("device_brand"),
            row.getAs[String]("device_model"),
            row.getAs[String]("view_class"),
            row.getAs[String]("view_path"),
            row.getAs[String]("alternate_info"),
            row.getAs[String]("first_app_version"),
            row.getAs[String]("service_name"),
            row.getAs[String]("tag8"),
            row.getAs[String]("tag9"),
            row.getAs[String]("component_tag")
          ))
        }
      }
      resList.iterator
    })
    resRdd
  }


  /**
   * @Description 计算session_end_time,分两张情况:
   *             1.如果是低版本数据,session有拆分记录,则取session_id的首条记录时间以及末次记录时间作为session起始结束时间
   *             2.如果是高版本数据,session无拆分记录,则取session_id对应的pseudo_session首次以及末次出现的记录时间作为session起止时间
   * @param sourceDF 源数据
   * @return
   */
  def getSessionTail(sourceDF: DataFrame, sparkSession: SparkSession): DataFrame = {
    sourceDF.createOrReplaceTempView("select_res_table")
    val sourceSql =
      """
        |select session_id,min(created_time) session_begin_time,max(created_time) session_end_time from refer_result_table
        | group by  session_id
        |""".stripMargin
    val firstSessionEndDF = sparkSession.sql(sourceSql)
    println("firstSessionEndDF.show()=========")
    firstSessionEndDF
  }
  def loadData(dataFrame: DataFrame, sparkSession: SparkSession, partitionDay: String,index:Integer,count:Integer): Unit = {
    dataFrame.createOrReplaceTempView("result_view")
    var insertSql = "insert overwrite"
    if(index!=1){
      insertSql = "insert into"
    }
    //补充新生session超过10个之后的session的session_begin_time与session_end_time
    val HIVE_TABLE4 = "pica_dw.dw_fact_log_session_term_tmp"
    val loadDataSql =
      s"""
         |${insertSql} table ${MyConfigSession.HIVE_TABLE4} partition(created_day='${partitionDay}')
         | select concat(regexp_replace( '${partitionDay}','-','') ,cast( (row_number() over(partition by 1 order by created_time)  )+${count}  as string)) id,
         | t1.session_id,device_token,user_id,user_token,mobile,menu_code,action_code,position,label_value,label_class,module_class1,module_class2,action_type,
         | cast(menu_begin_time as string),   cast(menu_end_time as string),  menu_time_diff,
         | action_step,device_type,device_brand,device_model, app_version, net_type ,created_time,date_time,
         | cast(session_begin_time as string) ,  cast(session_end_time as string), session_time_diff,
         | pre_session_id, user_identity_id,view_class,view_path,alternate_info,first_app_version,service_name,tag8,tag9,component_tag
         | from result_view t1
         """.stripMargin
    sparkSession.sql(loadDataSql)
  }
  def strToInt(str: String): Int = {
    val regex = """([0-9]+)""".r
    val res = str match{
      case regex(num) => num
      case _ => "0"
    }
    val resInt = Integer.parseInt(res)
    resInt
  }

}
case class SessionTerm(pseudo_session: String,
                       user_id: Integer,
                       mobile: String,
                       device_token: String,
                       user_token: String,
                       view_class: String,
                       view_path: String,
                       action: String,
                       action_type: String,
                       component_tag: String,
                       menu_code: String,
                       menu_code_new: String,
                       menu_code_offset:String,
                       action_code: String,
                       position: String,
                       label_value: String,
                       label_class: String,
                       module_class1: String,
                       module_class2: String,
                       app_version: String,
                       device_type: String,
                       device_brand: String,
                       device_model: String,
                       net_type: String,
                       created_time: Long,
                       date_time: String,
                       alternate_info: String,
                       first_app_version: String,
                       service_name: String,
                       tag8: String, tag9: String,
                       is_need:Boolean)
