package project


import java.util

import org.apache.spark.sql.catalog.Catalog
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import tool.sqlUtils
import tool.getmysqldf
import tool.savefile
import tool.readfile
import org.apache.flink.api.java.utils.ParameterTool
object jdbctohive{
  def apply(parameterTool: ParameterTool): jdbctohive = new jdbctohive(parameterTool)

  def main(args: Array[String]): Unit = {
    if (args.length==0){
      println(
        """
          |欢迎使用本程序
          |参数详情 mysql hive
          |-------------------------mysql
          |url 例子 ： jdbc:mysql://bigdata2:3306/try
          |user 例子 ： root
          |password 例子 ： liuzihan010616
          |tablename => 支持谓词下压  例子 ： emp 或者 select * from emp 等
          |driver => com.mysql.jdbc.Driver
          |---------------------------hive
          |mode模式 overwrite append 等
          |hive中的table 例子 bigdata.emp
          |可选参数 分区字段 自动开启的是动态分区 例子 deptno
          |分区字段 [字段值] [标志位]：代表是不是只更新这一个分区的数据
          |jdbc:mysql://bigdata2:3306/try root liuzihan010616 "select * from emp " com.mysql.jdbc.Driver append default.tmp deptno,sal,test,re 999,888
          |""".stripMargin)
    }
    val tool = ParameterTool.fromArgs(args)
    jdbctohive(tool).excute(args)
  }
}




class jdbctohive(parameterTool: ParameterTool) {
  System.setProperty("HADOOP_USER_NAME","hadoop")
  val spark = SparkSession.builder().appName("sqoop").master("local[4]").enableHiveSupport().getOrCreate()
  spark.sparkContext.setCheckpointDir("/tmp/checkpoint")

  val getmysqldf = new readfile
  val sqlUtils = new sqlUtils
  val saveFile = new savefile
  private val catalog: Catalog = spark.catalog
  var changecolunm = false
  import spark.implicits._
  import org.apache.spark.sql.functions._

  val url = parameterTool.getRequired("url")
  val user = parameterTool.getRequired("user")
  val password = parameterTool.getRequired("password")
  val table = parameterTool.getRequired("table")
  val driver = parameterTool.getRequired("driver")
  val mode = parameterTool.getRequired("mode")
  val hivetable = parameterTool.getRequired("hivetable")
  val hivepartition = parameterTool.get("hivepartition",null)
  val partitionValues = parameterTool.get("partitionValues")
  val insertpartition = parameterTool.get("insertpartition")


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


    // 获取jdbc的df
    val mysqlconnect = getmysqldf.getmysqldataframe(spark, url, user, password, table , driver)
    // 验证指示
    mysqlconnect.show()
    // 生成hive参数数组
//    var hiveconf = new Array[String](args.length-5)
//    hiveconf = util.Arrays.copyOfRange(args, 5, args.length)
    //hiveconf.foreach(println(_))
    jdbctohive(args.length,catalog,mysqlconnect)
    spark.stop()
  }



  def changecolnums(int: Int,resourcesql:DataFrame) ={
    var finallyresult:Dataset[Row] = null // 最终结果集
    var frame:DataFrame = null // 中间变量
    val strings2 = hivepartition.split(",")
    var hiveconclumns = spark.table(hivetable).columns // hive的列数
    //hiveconclumns.foreach(println((_))) // 验证hive的列数
    var mysqlconnect:DataFrame = resourcesql // 设置数据源的resource

    // 判断分区字段在不在jdbc的数据里，如果不在，则在jdbc的数据源中先添加上分区字段
    var strings1:Array[String] =null
    if (int > 8 && partitionValues != null){
      strings1 = partitionValues.split(",")
    }
    var flagtmp:Int = 0;
    for (elem <- strings2){
      if (!mysqlconnect.columns.contains(elem)){
        println(elem)
        println(strings1(flagtmp))
        mysqlconnect = mysqlconnect.withColumn(elem,lit(strings1(flagtmp)))
        flagtmp = flagtmp + 1
        mysqlconnect.show()
      }
    }



    val jdbcconclumns = mysqlconnect.columns // jdbc的列数


    var jdbcoldsource:Dataset[Row] = null // 源数据库的数据 checkpoint是为了破坏数据均衡，以后能编写变读取

    if (int == 10){
      hivepartition.split(",")(0) match {
        case "" => {
          println("-------------------------无操作")
        }
        case _ => {
          hivepartition.split(",").length match {
            case 1 =>
            {
              jdbcoldsource = spark.sql(
                s"""
                   |select * from ${hivetable} where ${hivepartition} != ${partitionValues}
                   |""".stripMargin).checkpoint()
            }
            case _ =>
            {
              var tmpstring:String  = null
              var flag:Int = 0
                 val flagvalue = partitionValues.split(",")
                  for (elem <- hivepartition.split(",")){
                 if (elem == hivepartition.split(",")(hivepartition.split(",").length-1)){
                   tmpstring = tmpstring + elem + "!=" + flagvalue(flag)
                 }else{
                   tmpstring = tmpstring + elem + "!=" + flagvalue(flag) + "and"
                 }
                    flag = flag + 1
              }
              jdbcoldsource = spark.sql(
                s"""
                   |select * from ${hivetable} where ${tmpstring}
                   |""".stripMargin).checkpoint()
            }
          }
        }
      }


    }else{
      jdbcoldsource =  spark.sql(
        s"""
           |select * from ${hivetable}
           |""".stripMargin).checkpoint()
    }

    var existcolunms: Array[String] = null  // 设置hive或者mysql的额外列
    var resultdf: DataFrame = jdbcoldsource // 获取hive的数据原始数据

    // 判断是hive的列多，还是数据源的列数多
    if (hiveconclumns.length >= jdbcconclumns.length){
      // 判断额外列的存在
      existcolunms= hiveconclumns.filter(hivecol => {
        val bool = jdbcconclumns.map(jdbccol => {
          jdbccol == hivecol
        }).contains(true)
        !bool
      })
      // 判断两个列数是不是相等
        if (existcolunms.isEmpty) {
          frame = mysqlconnect.selectExpr(hiveconclumns: _*)
          frame
        }else{
        // 列数不相等的时候让列数少的加列
        resultdf = mysqlconnect
        for (elem <- existcolunms){
          resultdf = resultdf.withColumn(elem, lit(null))
        }
        // 对字段进行排序 ， 让分区数据的分区字段在最后一列
        frame = resultdf.selectExpr(hiveconclumns: _*)
        // 验证数据
        frame.show()
        // 整合历史数据
        finallyresult = jdbcoldsource.union(frame)
        // 验证数据
        finallyresult.show()
        changecolunm = true
        finallyresult
      }
    }else{
      // 数据的列多
      existcolunms= jdbcconclumns.filter(jdbccol => {
        val bool = hiveconclumns.map(hivecol => {
          jdbccol == hivecol
        }).contains(true)
        !bool
      })

      if (existcolunms.isEmpty) {
        frame = mysqlconnect.selectExpr(hiveconclumns: _*)
        frame
      }else{
        for (elem <- existcolunms){
          resultdf = resultdf.withColumn(elem, lit(null))
        }
        frame = resultdf.selectExpr(jdbcconclumns: _*)
        finallyresult = frame.union(mysqlconnect)
        changecolunm = true
        finallyresult
      }
    }
  }






  def jdbctohive(int: Int,catalog: Catalog,mysqlconnect: DataFrame)={
    // 分割字符串获取hive的 表和数据库
    val hivedbandtables = hivetable.split("\\.")
    val hivepart = hivepartition.split(",")
    hivepart.foreach(println(_))
// catalog的方法 获取表存不存在的方法
//    catalog.listTables(strings(0)).show()
//    val empty = catalog.listTables(strings(0)).filter(x => {
//      x.name == strings(1)
//    }).isEmpty
    val empty = catalog.tableExists(hivedbandtables(0),hivedbandtables(1))
//-----------------------------------------------------------------------------
//    sql的方法
//    val empty1 = spark.sql(
//      """
//        |show tables in hivedb
//        |""".stripMargin).filter("tableName = 'hivetablename'").isEmpty
// --------------------------------------------------------------------------


    // 判断列数是不是相等
    var frameresult:DataFrame = mysqlconnect
    // 先判断表存不存在 ，因为判断列数的方法要求表存在
      empty match {
          // 表不存在
      case false => {
        // 判断输入的变量个数执行 判断分区表还是普通表
        if (int > 7) {
          println("-----------------分区表")
          // 判断分区的参数在不在列中 如果不在 ，则加上 ，在的话就自动往下走
          var hivepartval:Array[String] =null
            if (int > 8 && partitionValues != null){
            hivepartval = partitionValues.split(",")
          }
          var flagtmp:Int = 0;
          for (elem <- hivepart){
            if (!mysqlconnect.columns.contains(elem)){
              println(elem)
              println(hivepartval(flagtmp))
              frameresult = frameresult.withColumn(elem,lit(hivepartval(flagtmp)))
              flagtmp = flagtmp + 1
            }
          }
        }else{
          println("-----------普通表")
          frameresult = mysqlconnect
          mysqlconnect.show()
        }
      }

      case true => {
        // 表存在
        // 判断是不是分区表
        frameresult = changecolnums(int, mysqlconnect)
//        if (args.length > 7) {
//          println("-----------------分区表")
//          if (!mysqlconnect.columns.contains(args(7))){
//            frameresult = changecolnums(args, hiveconf, mysqlconnect)
//          }
//        }else{
//          println("-----------普通表")
//          frameresult = mysqlconnect
//        }
        frameresult.show()}
    }









    spark.conf.set("hive.exec.dynamic.partition.mode","nonstrict")
    spark.conf.set("hive.exec.dynamic.partition","true")
    spark.conf.set("spark.sql.parquet.writeLegacyFormat", "true")
    println(empty)
    saveFile.savetohiveapi(spark,empty,frameresult,hivetable,mode,hivepartition,changecolunm)
}

}

package sparkfirst

import org.apache.spark.sql.SparkSession
import tool.savefile
import tool.sqlUtils
import org.apache.spark.sql.functions._
object test {
  val spark = SparkSession.builder().appName("Sparksql01").master("local[4]").enableHiveSupport().getOrCreate()
  private val savefile = new savefile
  private val utils = new sqlUtils
  def main(args: Array[String]): Unit = {
    val df = spark.read.format("JDBC")
      .option("url","jdbc:mysql://bigdata2:3306/try")
      .option("dbtable", "emp")
      .option("user", "root")
      .option("password", "liuzihan010616")
      .load()
    df.select("sal").tail(1).foreach(println(_))
    println(df.select("sal").tail(1)(0)(0))
    df.show()


    var str:String = null
    val bool = spark.catalog.tableExists("default.tmp")
    if (bool){
      spark.sql(
        s"""
          |drop table default.tmp
          |""".stripMargin)
      str = utils.mkcreatesql(df, "default.tmp", "text", "','","deptno,hiredate")
      utils.checksql(spark,str)
    }else{
      str = utils.mkcreatesql(df, "default.tmp", "text", "','","deptno,hiredate")
      utils.checksql(spark,str)
    }

    spark.conf.set("hive.exec.dynamic.partition.mode","nonstrict")
    spark.conf.set("hive.exec.dynamic.partition","true")
    val frame = df.withColumn("ee", lit("aaa"))
    utils.insertmake(spark,df,"default.tmp","','","deptno,hiredate")
    utils.changecolunms(spark,frame,"default.tmp")
    utils.insertmake(spark,frame,"default.tmp","','","deptno,hiredate")


  }
}

    saveFile.savetohiveapi(spark,empty,frameresult,hivetable,mode,hivepartition,changecolunm,fileformated,codec)
------------------------------------------------------------------------------------------------------------------
def savetohiveapi(sparkSession: SparkSession,boolean: Boolean,spark: DataFrame,hivetable:String,mode:String,hivepartition:String,changecolnums:Boolean,fileformat:String,codec:String) = {




    if (!boolean){
      if (hivepartition != null){
        spark.write.partitionBy(hivepartition.split(","):_*).option("fileFormat",fileformat).option("compression",codec).mode(mode).format("hive").saveAsTable(hivetable)
      }else {

        println(hivetable)
        println(hivepartition)
        spark.write.option("fileFormat",fileformat).option("compression",codec).mode(mode).format("hive").saveAsTable(hivetable)
      }

    }else{
      changecolnums match {
        case true =>  {
          if (hivepartition != null){
            if(sparkSession.table(hivetable).columns.length != spark.columns.length){
             sparkSession.sql(
               s"""
                 |drop table ${hivetable}
                 |""".stripMargin)
            }
            spark.write.partitionBy(hivepartition.split(","):_*).option("fileFormat",fileformat).option("compression",codec).mode(mode).format("hive").saveAsTable(hivetable)
          }else {
            spark.write.option("fileFormat",fileformat).option("compression",codec).mode(mode).format("hive").saveAsTable(hivetable)
          }
        }
        case false => spark.write.option("fileFormat",fileformat).option("compression",codec).mode(mode).format("hive").insertInto(hivetable)
      }

      spark.show()
      println(spark.count())

    }
  }

def savetojdbc(spark: SparkSession,df: DataFrame, url:String,user:String,password:String,dbtable:String,driver:String,mideng:String,mode:String)={
    val map = Map("url" -> url,
      "user" -> user,
      "password" -> password,
      "dbtable" -> dbtable,
      "driver"-> driver)

//    df.write.mode(string(2)).format(string(1)).options(map).save()
    // -------------------------------------幂等性
    val connection = jdbcconnect.getconncet(driver,url,user,password)
    try{
      val bool = connection.createStatement().executeQuery(s"show tables like '${dbtable}'").next()
      if (!bool){
        throw new NullPointerException(s"写入的结果表${dbtable} 尚未创建！！！")
      }else{
        var flag:Any = null
        val flagbool = mysqldf.getmyqsldffromMap(spark, map).select(mideng).isEmpty
        if (!flagbool){
         flag = mysqldf.getmyqsldffromMap(spark, map).select(mideng).tail(1)(0)(0)
        }
        val tmpresult = df.select(mideng).filter(line => {
          line.getString(0) != flag
        })

        if (df.isEmpty){
          println("数据集为空")
        }else{
          if (tmpresult.isEmpty){
            println("你的数据已经插入过")
            df.show(false)
          }else {
//            df.show()
//            println(df.count())
            //val insertresult = tmpresult.join(df, string(5))
            tmpresult.show()
            println(tmpresult.count())
//            insertresult.show()
//            println(insertresult.count())
            tmpresult.write.mode(mode).format("jdbc").options(map).save()
          }
        }
      }
    }finally {
      connection.close()
    }


  }

Apache Flink is a framework and distributed processing engine for stateful computations over unbounded and bounded data streams. Flink has been designed to run in all common cluster environments, perform computations at in-memory speed and at any scale.

Here, we explain important aspects of Flink’s architecture.

SELECT userId, COUNT(*)
FROM clicks
GROUP BY SESSION(clicktime, INTERVAL '30' MINUTE), userId