优化session_pref与term

src/main/scala/com/session/SessionProcessTerm.scala

@@ -43,8 +43,12 @@ object SessionProcessTerm {
 		""".stripMargin
     //设置同步数据的批次号,格式是2019-09-12
     var scnData: String = DateUtils.getYesterdayDate
+    var condition = " 1=1"
     if (args.length >= 1) {
       scnData = args(0)
+      if(args.length > 1){
+        condition = args(1)
+      }
     }
     println(s"scnData=${scnData}")
     //设置任务开始时间,格式是2019-09-12 14:03:30
@@ -68,8 +72,8 @@ object SessionProcessTerm {
            | from ${MyConfigSession.HIVE_TABLE0}
            | where servicename='trace2' and action!='ACTION_EQUIP_INFO'
            | and (case when ((view_class like '%YunqueApp%' and action!='ACTION_HEART_BEAT') or LENGTH(view_class)<=3 or view_class='YQGuidePageViewVC') then '2' else '1' end)='1'
-           |   and created_day='${scnData}'
-           |""".stripMargin //  and pseudo_session='3b3cec3b-2305-4e3a-b690-843e2f666c69'
+           |   and created_day='${scnData}' and ${condition}
+           |""".stripMargin
       val sourceDF: DataFrame = sparkSession.sql(SOURCE_SQL_TERM )
       println("sourceDF.show==================")
       sourceDF.printSchema()
@@ -86,13 +90,15 @@ object SessionProcessTerm {
       println("selectDF.show========")
       selectDF.printSchema()
       println("selectDF.count=========",selectDF.count())
-      val conditionGroup = List("<='4' ","between '5' and '9'",">'9'")
+//      val conditionGroup = List("<='4' ","between '5' and '9'",">'9'")
+      val conditionGroup = List("='0'","='1'","='2'","='3'","='4'","='5'","='6'","='7'","='8'","='9'",
+        "='a'","='b'","='c'","='d'","='e'","='f'")
       var dataCount = 0
       var index = 0
       selectDF.persist(StorageLevel.MEMORY_AND_DISK_SER)
       for(condition <- conditionGroup){
         index += 1
-        val slideDF = selectDF.where(s" SUBSTRING(pseudo_session,1,1)  ${condition}")
+        val slideDF = selectDF.where(s" SUBSTRING(pseudo_session,2,1)  ${condition}").repartition(100)
         println(s"-----------------------------------compute refer columns,condition=${condition}-----------------------------------------")
         val referResultRdd = sessionProcessTerm.getReferColumns(slideDF)
         val referResultDF: DataFrame = sparkSession.createDataFrame(referResultRdd, StructType(
@@ -118,7 +124,7 @@ object SessionProcessTerm {
         println("referResultDF.show()============'")
         referResultDF.printSchema()
 //        referResultDF.where("action_type='ACTION_VIEW'").show(100,true)
-        referResultDF.repartition(100).persist(StorageLevel.MEMORY_AND_DISK_SER).createOrReplaceTempView("refer_result_table")
+        referResultDF.persist(StorageLevel.MEMORY_AND_DISK_SER).createOrReplaceTempView("refer_result_table")
         println("-----------------------------------compute menu_code term-----------------------------------------")
         val getMenuTermSql =
           """
@@ -172,7 +178,7 @@ object SessionProcessTerm {
              |(cast(c.session_end_time as bigint)-cast(c.session_begin_time as bigint))/1000 session_time_diff,
              |a.refer_session_id
              |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
+             |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
@@ -191,11 +197,6 @@ object SessionProcessTerm {
         println(s"${condition}的结果==${resCount}")
         dataCount += resCount
       }
-      val fields = List("id","session_id","device_token","user_id","mobile","menu_code","menu_begin_time","menu_end_time","menu_time_diff","action_type",
-        "action_code","position","label_value","refer_menu_code","refer_action_code","refer_position","refer_label_value","refer_action_type","action_step",
-        "device_type","app_version","created_time","date_time","session_begin_time","session_end_time","session_time_diff","refer_session_id")
-//      sparkSession.sql(s"insert overwrite table ${MyConfigSession.HIVE_TABLE4} partition(created_day='${scnData}') " +
-//        s"select ${fields.mkString(",")} from ${MyConfigSession.HIVE_TABLE4} where created_day='${scnData}'")
       println("----------------------------------update task record table---------------------------------------")
       //任务执行成功,更新 Mysql record 配置表
       val updateSQL: String =

src/main/scala/com/utils/UseUtil.scala

@@ -88,6 +88,8 @@ object UseUtil {
 		conf.set("spark.reducer.maxSizeInFlight", "96m")
 		//设置字符串
 		conf.set("spark.debug.maxToStringFields","100")
+		//启用自动设置 Shuffle Reducer,默认false
+		conf.set("spark.sql.adaptive.enabled","true")
 	}