• SparkSql批量插入或更新，保存数据到Mysql中

    在sparksql 中,保存数据到数据,只有 Append , Overwrite , ErrorIfExists, Ignore 四种模式,不满足项目需求 ,此处大概说一下我们需求，当业务库有数据发生变化，需要更新、插入、删除数仓中ods层的数据，因此需要改造源码。

    现依据 spark save 源码,进行进一步的改造, 批量保存数据,存在则更新 不存在 则插入

    1. import com.sun.corba.se.impl.activation.ServerMain.logError
    2. import org.apache.spark.SparkContext
    3. import org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils.getCommonJDBCType
    4. import org.apache.spark.sql.jdbc.{JdbcDialect, JdbcDialects, JdbcType}
    5. import org.apache.spark.sql.types._
    6. import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}
    7.  
    8. import java.sql.{Connection, DriverManager, PreparedStatement}
    9. import java.util.Properties
    10.  
    11. object TestInsertOrUpdateMysql {
    12.  
    13.   val url: String = "jdbc:mysql://192.168.1.1:3306/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false&allowMultiQueries=true&autoReconnect=true&failOverReadOnly=false"
    14.   val driver: String = "com.mysql.jdbc.Driver"
    15.   val user: String = "123"
    16.   val password: String = "123"
    17.   val sql: String = "select * from testserver "
    18.   val table: String = "testinsertorupdate"
    19.  
    20.   def main(args: Array[String]): Unit = {
    21.     val spark = SparkSession.builder()
    22.       .master("local[*]")
    23.       .appName("testSqlServer").getOrCreate()
    24.  
    25.     val dbtable = "(" + sql + ") AS Temp"
    26.  
    27.     val jdbcDF = spark.read
    28.       .format("jdbc")
    29.       .option("driver", driver)
    30.       .option("user", user)
    31.       .option("password", password)
    32.       .option("url", url)
    33.       .option("dbtable", dbtable)
    34.       .load()
    35.     jdbcDF.show()
    36.  
    37.     //普通写入数据库
    38.     //commonWrite(jdbcDF)
    39.  
    40.     //saveorupdate
    41.     insertOrUpdateToMysql("id", jdbcDF, spark)
    42.  
    43.     println("======================程序结束======================")
    44.   }

    1.先看一下普通的插入怎么写的

    1. def commonWrite(jdbcDF: DataFrame): Unit = {
    2.     val properties = new Properties()
    3.     properties.put("user", user)
    4.     properties.put("password", password)
    5.     properties.put("driver", driver)
    6.     jdbcDF.write.mode(SaveMode.Append).jdbc(url, table, properties)
    7.   }

    这种方式比较局限，只能做一些简单的插入（追加或覆盖等SaveMode.Append）

    那么新的写法是什么呢，首先写出mysql的更新或插入的语法规则：

    1. INSERT INTO t_name ( c1, c2, c3 )
    2. VALUES
    3. 	( 1, '1', '1') 
    4. 	ON DUPLICATE KEY UPDATE
    5. 	c2 = '2';

    需要注意的是一定要有主键，没主键没法更新；

    2.看一下insertorupdate的写法

    1. //写入数据库，批量插入 或更新 数据 ,该方法 借鉴Spark.write.save() 源码
    2.   // 规则如下：
    3.   //没有关键主键字段即为插入，有即为更新
    4.   def insertOrUpdateToMysql(primaryKey: String, jdbcDF: DataFrame, spark: SparkSession): Unit = {
    5.     val sc: SparkContext = spark.sparkContext
    6.     spark.conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    7.     //1.加载驱动程序
    8.     Class.forName(driver);
    9.     //2. 获得数据库连接
    10.     val conn: Connection = DriverManager.getConnection(url, user, password);
    11.  
    12.     val tableSchema = jdbcDF.schema
    13.     val columns = tableSchema.fields.map(x => x.name).mkString(",")
    14.     val placeholders = tableSchema.fields.map(_ => "?").mkString(",")
    15.     val sql = s"INSERT INTO $table ($columns) VALUES ($placeholders) on duplicate key update "
    16.     val update = tableSchema.fields.map(x =>
    17.       x.name.toString + "=?"
    18.     ).mkString(",")
    19.  
    20.     //ON DUPLICATE KEY UPDATE
    21.     //on conflict($primaryKey) do update set
    22.     val realsql = sql.concat(update)
    23.     conn.setAutoCommit(false)
    24.     val dialect = JdbcDialects.get(conn.getMetaData.getURL)
    25.     val broad_ps = sc.broadcast(conn.prepareStatement(realsql))
    26.     val numFields = tableSchema.fields.length * 2
    27.     //调用spark中自带的函数,获取属性字段与字段类型
    28.     val nullTypes = tableSchema.fields.map(f => getJdbcType(f.dataType, dialect).jdbcNullType)
    29.     val setters = tableSchema.fields.map(f => makeSetter(conn, f.dataType))
    30.  
    31.     var rowCount = 0
    32.     val batchSize = 2000
    33.     val updateindex = numFields / 2
    34.     try {
    35.       jdbcDF.foreachPartition(iterator => {
    36.         //遍历批量提交
    37.         val ps = broad_ps.value
    38.         try {
    39.           while (iterator.hasNext) {
    40.             val row = iterator.next()
    41.             var i = 0
    42.             while (i < numFields) {
    43.               i < updateindex match {
    44.                 case true => {
    45.                   if (row.isNullAt(i)) {
    46.                     ps.setNull(i + 1, nullTypes(i))
    47.                   } else {
    48.                     setters(i).apply(ps, row, i, 0)
    49.                   }
    50.                 }
    51.                 case false => {
    52.                   if (row.isNullAt(i - updateindex)) {
    53.                     ps.setNull(i + 1, nullTypes(i - updateindex))
    54.                   } else {
    55.                     setters(i - updateindex).apply(ps, row, i, updateindex)
    56.                   }
    57.                 }
    58.               }
    59.               i = i + 1
    60.             }
    61.             ps.addBatch()
    62.             rowCount += 1
    63.             if (rowCount % batchSize == 0) {
    64.               ps.executeBatch()
    65.               rowCount = 0
    66.             }
    67.           }
    68.           if (rowCount > 0) {
    69.             ps.executeBatch()
    70.           }
    71.         } finally {
    72.           ps.close()
    73.         }
    74.       })
    75.       conn.commit()
    76.     } catch {
    77.       case e: Exception =>
    78.         logError("Error in execution of insert. " + e.getMessage)
    79.         conn.rollback()
    80.       // insertError(connectionPool("OuCloud_ODS"),"insertOrUpdateToPgsql",e.getMessage)
    81.     } finally {
    82.       conn.close()
    83.     }
    84.  
    85.   }

    几个源码包

    1. private def getJdbcType(dt: DataType, dialect: JdbcDialect): JdbcType = {
    2.     dialect.getJDBCType(dt).orElse(getCommonJDBCType(dt)).getOrElse(
    3.       throw new IllegalArgumentException(s"Can't get JDBC type for ${dt.catalogString}"))
    4.   }
    5.  
    6.   private type JDBCValueSetter_add = (PreparedStatement, Row, Int, Int) => Unit
    7.  
    8.   private def makeSetter(conn: Connection, dataType: DataType): JDBCValueSetter_add = dataType match {
    9.     case IntegerType =>
    10.       (stmt: PreparedStatement, row: Row, pos: Int, currentpos: Int) =>
    11.         stmt.setInt(pos + 1, row.getInt(pos - currentpos))
    12.  
    13.     case LongType =>
    14.       (stmt: PreparedStatement, row: Row, pos: Int, currentpos: Int) =>
    15.         stmt.setLong(pos + 1, row.getLong(pos - currentpos))
    16.  
    17.     case DoubleType =>
    18.       (stmt: PreparedStatement, row: Row, pos: Int, currentpos: Int) =>
    19.         stmt.setDouble(pos + 1, row.getDouble(pos - currentpos))
    20.  
    21.     case FloatType =>
    22.       (stmt: PreparedStatement, row: Row, pos: Int, currentpos: Int) =>
    23.         stmt.setFloat(pos + 1, row.getFloat(pos - currentpos))
    24.  
    25.     case ShortType =>
    26.       (stmt: PreparedStatement, row: Row, pos: Int, currentpos: Int) =>
    27.         stmt.setInt(pos + 1, row.getShort(pos - currentpos))
    28.  
    29.     case ByteType =>
    30.       (stmt: PreparedStatement, row: Row, pos: Int, currentpos: Int) =>
    31.         stmt.setInt(pos + 1, row.getByte(pos - currentpos))
    32.  
    33.     case BooleanType =>
    34.       (stmt: PreparedStatement, row: Row, pos: Int, currentpos: Int) =>
    35.         stmt.setBoolean(pos + 1, row.getBoolean(pos - currentpos))
    36.  
    37.     case StringType =>
    38.       (stmt: PreparedStatement, row: Row, pos: Int, currentpos: Int) =>
    39.         stmt.setString(pos + 1, row.getString(pos - currentpos))
    40.  
    41.     case BinaryType =>
    42.       (stmt: PreparedStatement, row: Row, pos: Int, currentpos: Int) =>
    43.         stmt.setBytes(pos + 1, row.getAs[Array[Byte]](pos - currentpos))
    44.  
    45.     case TimestampType =>
    46.       (stmt: PreparedStatement, row: Row, pos: Int, currentpos: Int) =>
    47.         stmt.setTimestamp(pos + 1, row.getAs[java.sql.Timestamp](pos - currentpos))
    48.  
    49.     case DateType =>
    50.       (stmt: PreparedStatement, row: Row, pos: Int, currentpos: Int) =>
    51.         stmt.setDate(pos + 1, row.getAs[java.sql.Date](pos - currentpos))
    52.  
    53.     case t: DecimalType =>
    54.       (stmt: PreparedStatement, row: Row, pos: Int, currentpos: Int) =>
    55.         stmt.setBigDecimal(pos + 1, row.getDecimal(pos - currentpos))
    56.     case _ =>
    57.       (stmt: PreparedStatement, row: Row, pos: Int, currentpos: Int) =>
    58.         throw new IllegalArgumentException(
    59.           s"Can't translate non-null value for field $pos")
    60.   }

    这里面有个属性比较关键我列出来，不加会报错 Exception in thread "main" java.io.NotSerializableException: com.mysql.jdbc.JDBC42PreparedStatement：

    spark.conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")

    添加下这个配置 这个第三方序列化 用默认的javaSerializer 不行

    结果如下，这块有一些坑，有需要的朋友，我们可以交流

    附postgreesql的更新或插入语法：

    1. INSERT INTO test_001 ( c1, c2, c3 )
    2. VALUES( ?, ?, ? )
    3. ON conflict ( ID ) DO
    4. UPDATE SET c1=?,c2 = ?,c3 = ?;

    MySQL的on duplicate key update 的使用_厄尔尼诺的夏天的博客-CSDN博客 

    博主qq：907044657，欢迎大家一起交流学习，有问题请指出，转载麻烦注明出处，多谢啦

  • 相关阅读:
    axios拦截器
    UI设计师的发展前景是否超越了平面设计？
    用冒泡排序完成库函数qsort的作用
    可靠的自托管「GitHub 热点速览 v.22.37」
    28 行为型模式-中介者模式
    服务器常用端口号总结
    CSS变量 var()的用法
    【luogu P3295】萌萌哒（并查集）（倍增）
    「笔试刷题」：腐烂的苹果
    Spring Boot访问静态资源
  • 原文地址：https://blog.csdn.net/Alex_81D/article/details/125893271