Spark insertOrUpdate MySQL数据

旧城等待， 2021-09-24 02:50 362阅读 0赞

> 需求：使用 Spark 将 Hive 数据同步到 MySQL，MySQL表以其中三个字段作为唯一索引，索引不冲突的数据直接写入，冲突的数据对其中几个字段进行更新。

**主类**

import java.util.Properties
    
    import com.sm.conf.ConfigManager
    import com.sm.constants.Constants
    import com.sm.utils.{DateUtils, MySQLUtils}
    import org.apache.log4j.Level
    import org.apache.spark.sql.{SaveMode, SparkSession}
    import org.slf4j.LoggerFactory
    
    /**
      * 同步 Hive dws payment及sv_key retained 数据到MySQL
      *
      * create by LiuJinHe 2020/4/1
      */
    object ExportPayRetainedToMysql {
      private val warehouseLocation = "hdfs://cdh-slave01:9870/user/hive/warehouse"
      private var prop: Properties = _
      private var hiveDB: String = _
      private var database: String = _
      private var url: String = _
      private var hiveTable: String = _
      private var destTable: String = _
      private var logger: org.slf4j.Logger = _
      private var yesterday: String = _
      private var start: Long = 0L
      private var sqlStr: String = _
    
      def main(args: Array[String]): Unit = {
        logger = LoggerFactory.getLogger(this.getClass)
        org.apache.log4j.Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
        org.apache.log4j.Logger.getLogger("org.apache.spark").setLevel(Level.INFO)
        org.apache.log4j.Logger.getLogger("org.spark_project.jetty").setLevel(Level.WARN)
    
        val spark = initSparkSession
        start = System.currentTimeMillis()
    
        // 拉取数据日期,格式为2020-01-01,默认为前一天
        yesterday = DateUtils.getYesterdayDate
    
        if (args.length == 1) {
          yesterday = args(0)
        }
    
        prop = ConfigManager.load(Constants.PAY_RETAINED_DB)
        database = prop.getProperty("database")
        url = prop.getProperty("url")
    
        hiveDB = prop.getProperty("hive.database")
    
        hiveTable = prop.getProperty("hive.dws.user.payment.table")
        destTable = prop.getProperty("pay.table")
        exportDataToMySQL(spark, hiveDB, hiveTable, destTable)
    
        hiveTable = prop.getProperty("hive.dws.sv.key.retained.detail.table")
        destTable = prop.getProperty("retained.table")
        exportDataToMySQL(spark, hiveDB, hiveTable, destTable)
    
        spark.stop()
    
        val end = System.currentTimeMillis()
        logger.warn(s"========== 耗时： ${(end - start) / 1000} 秒 ==========")
      }
    
      def exportDataToMySQL(spark: SparkSession, hiveDB: String, hiveTable: String, destTable: String): Unit = {
        logger.warn(s"========== 开始导出 $yesterday $hiveTable 数据 ==========")
    
        if (destTable.equals("tb_remain_user_daily")) {
          sqlStr =
            s"""
               |select
               | core_account, cp_game_id, game_id, package_id, sv_key, retained_day as remain_status, 
               | date_format(add_date,'yyyyMMdd') as tday,
               | 0 as is_remove, current_date as send_date, now() as update_time
               |from $hiveDB.$hiveTable
            """.stripMargin
    
          val dataFrame = spark.sql(sqlStr)
    
          // update 字段
          val updateCol = Array("remain_status", "send_date", "update_time")
    
          MySQLUtils.insertOrUpdate(prop, database, destTable, dataFrame, updateCol)
        }
    
        logger.warn(s"========== 写入MySQL表 $destTable 成功! ==========")
    
      }
    
      def initSparkSession: SparkSession = SparkSession.builder()
        .appName(this.getClass.getSimpleName)
        .master(Constants.SPARK_YARN_CLIENT_MODE)
        .config("spark.sql.warehouse.dir", warehouseLocation)
        .config("hive.exec.dynamic.partition", "true")
        .config("hive.exec.dynamic.partition.mode", "nonstrict")
        .config("hive.exec.max.dynamic.partitions", 2000)
        .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        .config("spark.kryoserializer.buffer", "1024m")
        .config("spark.kryoserializer.buffer.max", "2046m")
        .config("spark.io.compression.codec", "snappy")
        .config("spark.sql.codegen", "true")
        .config("spark.sql.unsafe.enabled", "true")
        .config("spark.shuffle.manager", "tungsten-sort")
        .enableHiveSupport()
        .getOrCreate()
    }

**DBUtils  数据库连接池**

import java.sql.Connection
    import java.util.Properties
    
    import org.apache.commons.dbcp2.BasicDataSource
    
    
    /**
      * 线程池设置，获取线程
      *
      * create by LiuJinHe 2020/3/23
      */
    object DBUtils extends Serializable {
    
      /**
        * dbcp2
        */
      def getConnection(dataSource: BasicDataSource, prop:Properties): Connection ={
        dataSource.setDriverClassName(prop.getProperty("driver"))
        dataSource.setUrl(prop.getProperty("url"))
        dataSource.setUsername(prop.getProperty("username"))
        dataSource.setPassword(prop.getProperty("password"))
    
        dataSource.setInitialSize(10)
        dataSource.setMaxTotal(50)
        dataSource.setMaxIdle(5)
    
        val conn = dataSource.getConnection
        println("创建 dbcp2 连接池：" + conn)
        conn
      }
    }

**MySQLUtils **

import java.sql.{Connection, PreparedStatement}
    import java.util.Properties
    
    import org.apache.commons.dbcp2.BasicDataSource
    import org.apache.spark.sql.DataFrame
    
    /**
      * MySQL JDBC
      *
      * create by LiuJinHe 2020/4/1
      */
    object MySQLUtils {
      val dataSource: BasicDataSource = new BasicDataSource
    
      /**
        * 拼接 sql 语句
        */
      def getSql(database: String, table: String, cols: Array[String], updateCols: Array[String]): String = {
        // 构建 insert into table(field1,field2)
        val sqlStr = new StringBuffer()
        // 构建 values(?,?)
        val sqlValue = new StringBuffer
    
        sqlStr.append(s"insert into $database.$table(")
    
        for (i <- cols.indices) {
          val field = cols(i)
          sqlStr.append(s"$field,")
          sqlValue.append("?,")
        }
    
        sqlStr.replace(sqlStr.length - 1, sqlStr.length(), ") values(")
          .append(sqlValue.replace(sqlValue.length - 1, sqlValue.length(), ") on duplicate key update "))
    
        // 拼接更新字段
        for (field <- updateCols) {
          if (field.contains("remain_status"))
            sqlStr.append(s"$field = concat($field , ',',values($field)),")
          else
            sqlStr.append(s"$field = values($field),")
        }
        sqlStr.substring(0, sqlStr.length - 1)
      }
    
      /**
        * 写入或更新
        *
        * @param table      目标表
        * @param dateFrame  入库的 dataFrame
        * @param updateCols 更新的字段
        */
      def insertOrUpdate(prop: Properties, database: String, table: String, dateFrame: DataFrame, updateCols: Array[String]): Unit = {
        var nums: Array[Int] = Array()
        var count = 0
        val cols = dateFrame.columns
    
        val sql = getSql(database, table, cols, updateCols)
        println(s"======== 构建 sql ========\n$sql")
    
        dateFrame.repartition(1).foreachPartition(partitions => {
    
          val conn = DBUtils.getConnection(dataSource, prop)
          val ps = conn.prepareStatement(sql)
    
          try {
            conn.setAutoCommit(false)
    
            partitions.foreach(record => {
              // ps 赋值
              for (i <- cols.indices) {
                val value = record.get(i)
                ps.setObject(i + 1, value)
              }
              ps.addBatch()
              count += 1
    
              if (count % 10000 == 0) {
                nums = ps.executeBatch()
              }
            })
            conn.commit()
    
          } catch {
            case e: Exception => e.getMessage
          } finally {
            nums = ps.executeBatch()
            conn.commit()
    
            close(conn, ps)
          }
        })
      }
    
      def close(conn: Connection, ps: PreparedStatement): Unit = {
        if (conn != null) {
          conn.close()
        }
    
        if (ps != null) {
          ps.close()
        }
      }
    }