1）insert into test.table4(id,name) values('12141','xiaoming+') on duplicate key update name = VALUES(name)

2）这种语法 即是更新也是插入，具备两种功能。如果插入的数据表里面主键id是不存在的,那就不插入进去。如果是主键id是存在的,那就更新id的值,如果这个值是一样的,那就不会更新

package comm.myscala

import java.sql.{Connection, PreparedStatement}

import java.text.SimpleDateFormat

import java.util.{Date, Properties}

import org.apache.commons.dbcp2.BasicDataSource

import org.apache.spark.sql.DataFrame

object insertOrUpdateUtils extends Serializable {

/**

* 配置文件案例

*/

// val prop = new Properties()

// prop.put("username", "root") // 特备注意 有点人写user 有点人写 username,写法是不统一的

// prop.put("password", "a")

// prop.put("url", "jdbc:mysql://localhost:3306/test")

// prop.put("driver", "com.mysql.jdbc.Driver")

/**

* dbcp2连接池

*/

def getConnection(dataSource: BasicDataSource, prop:Properties): Connection ={

dataSource.setDriverClassName(prop.getProperty("driver"))

dataSource.setUrl(prop.getProperty("url"))

dataSource.setUsername(prop.getProperty("user"))

dataSource.setPassword(prop.getProperty("password"))

dataSource.setInitialSize(10)

dataSource.setMaxTotal(50)

dataSource.setMaxIdle(5)

val conn = dataSource.getConnection

// println("创建dbcp2连接池：" + conn)

conn

}

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(getDatetime + " insert into ..... ")

dateFrame.repartition(1).foreachPartition(partitions => {

val conn = insertOrUpdateUtils.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) { // 每到10000条 数据就进行写入 ，这样效率比较好

nums = ps.executeBatch()

}

})

conn.commit()

} catch {

case e: Exception => e.getMessage

} finally {

nums = ps.executeBatch()

conn.commit()

close(conn, ps)

// println("关闭连接池")

}

})

}

/**

* 关闭

*/

def close(conn: Connection, ps: PreparedStatement): Unit = {

if (conn != null) {

conn.close()

}

if (ps != null) {

ps.close()

}

}

/**

* 时间

* @return

*/

def getDatetime():String={

val now = new Date()

now.getTime

val dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

val result = dateFormat.format( now )

result

}

}