package com.aliyun.odps.spark.sparksql.profile

import com.alibaba.fastjson.JSONObject
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SparkSession}
import java.{sql, util}
import java.sql.{PreparedStatement, Timestamp}

import config.MyConfigSession
import utils.{DateUtils, MySqlUtil, StringUtils}

object Business_profile {
  def apply(): Business_profile = new Business_profile()
  def main(args: Array[String]): Unit = {
    //1.执行任务之前先往record表记录
//    val insertSQL: String =
//      s"""
//               			   |insert into ${MyConfigSession.DATA_BASE}.${MyConfigSession.JDBC_TABLE} (job_id,job_name,job_type,job_scn,status,start_time)
//               			   |values(0,'Business_profile','3',?,'0',?)
//		""".stripMargin
    //设置同步数据的批次号,格式是2019-09-12
    var scnData: String = DateUtils.getTodayDate
    var first_lable_id = ""
    var tbname = ""
    var list =

    if (args.length >= 1) {
      scnData = args(0)
      if (args.length > 1 && args(1) != "") {
        first_lable_id = args(1)
      }
      if (args.length > 2 && args(2) != "") {
        tbname = args(2)
      }
    }
    println(s"scnData=${scnData}")
    //设置任务开始时间,格式是2019-09-12 14:03:30
    val startTime: String = DateUtils.getTodayTime
    //存储SQL中的参数
    val insertArr: Array[String] = Array[String](scnData, startTime)
    //获取MYSQL连接
//    val connSql: sql.Connection = JDBCUtil.getConnection()
    //向 record 表插入数据
//    val flag: Int = JDBCUtil.insertRecord(connSql, insertSQL, insertArr)
    val business_profile: Business_profile = Business_profile()
    try {
      val sparkSession: SparkSession = business_profile.getSparkSession("Business_profile")

      var conditionGroup = List("between 7693 and 7718",
                                  "between 7719 and 8031",
                                  "between 8032 and 8083",
                                  "between 8084 and 8149",
                                  "between 8150 and 8195",
                                  "between 8196 and 8362",
                                   "between 8363 and 8377",
                                    "between 8378 and 9531",
                                    "between 9532 and 9568",
                                    "between 9569 and 9653"
      )
      var index = 0  //插入新增标识
      for(condition <- conditionGroup){
        index += 1
        println("查询标签的sql"+s"select * from ods_pica_bi_user_profile where sql_string <>'' and first_lable_id in (${first_lable_id}) and  third_label_id ${condition}    order by third_label_id ")
        val list_row: util.List[Row] = sparkSession.sql(
          s"""
             |
             |select * from ods_pica_bi_user_profile
             |where sql_string <>''
             |and first_lable_id in (${first_lable_id})
             |and  third_label_id ${condition}
             |order by third_label_id
             |""".stripMargin).collectAsList()

//      val profileListSQL=s"select * from user_profile where sql_string <>'' and first_lable_id in (${first_lable_id}) and  third_label_id ${condition}    order by third_label_id "
//      val MYSQL_URL = MyConfigSession.mysql_url_user_profile
//      val profileList: util.List[JSONObject] = MySqlUtil.queryList(profileListSQL,MYSQL_URL)
      import collection.JavaConverters._
      val profileListScala = list_row.asScala
      val tbbuffer = new StringBuilder
      val partitionDay = DateUtils.getTodayDate
      for(profile <- profileListScala){
       val first_lable_id= profile.getAs[String]("first_lable_id")
       val first_lable_name= profile.getAs[String]("first_lable_name")
       val secondary_label_id= profile.getAs[String]("secondary_label_id")
       val secondary_label_name =  profile.getAs[String]("secondary_label_name")
       val third_label_id = profile.getAs[String]("third_label_id")
       val third_label_name = profile.getAs[String]("third_label_name")
        val sql_string = profile.getAs[String]("sql_string")
        val Stringstr = s"""
             |select ${first_lable_id} as first_lable_id,'${first_lable_name}' as first_lable_name
             |       ,${secondary_label_id} as secondary_label_id,'${secondary_label_name}' as secondary_label_name
             |       ,${third_label_id} as third_label_id,'${third_label_name}' as third_label_name,A.user_id,A.value,A.acct_id,A.profile_id
             |from ${first_lable_id}_${secondary_label_id}_${third_label_id} A
             |""".stripMargin

        println("计算的sql拼接："+Stringstr)

        val df = sparkSession.sql(sql_string).repartition(400)
        df.createOrReplaceTempView(s"${first_lable_id}_${secondary_label_id}_${third_label_id}")
        sparkSession.sql(
            s"""
                |select ${first_lable_id} as first_lable_id,'${first_lable_name}' as first_lable_name
                |       ,${secondary_label_id} as secondary_label_id,'${secondary_label_name}' as secondary_label_name
                |       ,${third_label_id} as third_label_id,'${third_label_name}' as third_label_name,user_id,value,acct_id,profile_id
                |from ${first_lable_id}_${secondary_label_id}_${third_label_id}
                |""".stripMargin).createOrReplaceTempView(s"temp_${first_lable_id}_${secondary_label_id}_${third_label_id}")




        if(StringUtils.isEmpty(tbbuffer.toString())){
          var insertSql = "insert overwrite"
          if(index!=1){
            insertSql = "insert into"
          }
            tbbuffer.append(
              s"""
                 |${insertSql} table ${tbname}  partition(created_day='${scnData}')
                 | select user_id,cast(value as string),CAST(acct_id AS STRING),CAST(profile_id AS STRING),first_lable_id,first_lable_name,secondary_label_id,secondary_label_name,third_label_id,third_label_name
                 | FROM temp_${first_lable_id}_${secondary_label_id}_${third_label_id}
                 |""".stripMargin)
          }else{
            tbbuffer.append(
              s"""
                 |union all
                 |select user_id,cast(value as string),CAST(acct_id AS STRING),CAST(profile_id AS STRING),first_lable_id,first_lable_name,secondary_label_id,secondary_label_name,third_label_id,third_label_name
                 |FROM temp_${first_lable_id}_${secondary_label_id}_${third_label_id}
                 |""".stripMargin
            )
        }
      }
        println("------------------------------------------------开始最后的union拼接sql------------------------------------------------------------"+tbbuffer.toString())
        val resulteDf = sparkSession.sql(tbbuffer.toString())
        println("------------------------------------------------success------------------------------------------------------------"+tbbuffer.toString())

      }
      sparkSession.stop()


//      //任务执行成功,更新 Mysql record 配置表
//      val updateSQL: String =
//        s"""
//           |update ${MyConfigSession.JDBC_TABLE} set status=?,end_time=? where job_name='Business_profile' and start_time='${startTime}'
//				""".stripMargin
//      val upreSta: PreparedStatement = connSql.prepareStatement(updateSQL)
//      upreSta.setString(1, "1")
//      upreSta.setString(2, DateUtils.getTodayTime)
//
//      //更新表数据
//      upreSta.executeUpdate()
//      //关闭连接
////      JDBCUtil.close(connSql, upreSta)
      sparkSession.stop()

    } catch {
      case e: Exception => {
        println("-----------------------------------任务异常---------------------------------------------------")
        e.printStackTrace()
        throw new Exception

//        val exceptionSQL: String =
//          s"""
//             |update ${MyConfigSession.JDBC_TABLE} set status=?,exception=?,end_time=? where job_name='Business_profile' and start_time='${startTime}'
//					""".stripMargin
//        val errorArr = Array[String]("2", e.getMessage, DateUtils.getTodayTime)
//        JDBCUtil.insertRecord(connSql, exceptionSQL, errorArr)
//        connSql.close()

      }

    }
  }




}


class Business_profile {
  def getSparkSession(appName: String): SparkSession = {
//        val sparkSession = SparkSession
//          .builder()
//          .appName("SparkPi")
//          .config("spark.master", "local[1]") // 需设置spark.master为local[N]才能直接运行，N为并发数。
//          .config("spark.hadoop.odps.project.name", "pica")
//          .config("spark.hadoop.odps.access.id", "LTAI5tQEnb9VREcPdDAR32bg")
//          .config("spark.hadoop.odps.access.key", "RYazt1JIYZnRcEdlntarZ1Ov21zoNi")
//          .config("spark.hadoop.odps.end.point", "http://service.cn.maxcompute.aliyun.com/api")
//          .config("spark.sql.catalogImplementation", "odps")
//          .getOrCreate()

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

    sparkSession
  }
}
case class profile(
                             first_lable_id:Int,
                             first_lable_name:String,
                             secondary_label_id:Int,
                             secondary_label_name:String,
                             third_label_id:Int,
                             third_label_name:String,
                             load_date:Timestamp,
                             sql_string:String
                           )