1.读取/写入属性配置
Spark SQL支持以JDBC的方式读取或写入数据。用户可以使用Data Sources API将来自远程数据库的表作为 DataFrame 或 Spark SQL 临时视图进行加载。
也可以在数据源选项中指定JDBC连接属性，user(用户)和password(密码)通常作为登录数据源的连接属性提供。除了连接属性外，还支持的属性配置有：
<1> url：要连接的JDBC URL。源特定的连接属性可以在URL中指定，即在url后加上user和password
<2> dbtable：应该读取或者写入的JDBC表，可以使用SQL查询的FROM子句中有效的任何内容，比如，你可以使用括号中的子查询代替完整表，但是“dbtable”和“query”不可以同时使用。
<3> query：可以使用查询语句用来代替完整表，作为读取数据。
例如
spark.read.format("jdbc")
.option("url",jdbcUrl)
.option("query","select c1,c2 from t1")
.load()
使用query属性时，有一些限制
a、“dbtable”和“query”不可以同时使用
b、“query”和“partitionColumn”不可以同时使用
<4> driver：用于连接到此URL的JDBC driver程序的类名。
例如 MYSQL：com.mysql.jdbc.Driver
ORACLE：oracle.jdbc.OracleDriver
HIVE2：org.apache.hive.jdbc.HiveDriver
其他属性：
image.png
2.写入时的SaveMode
写入数据时可以配置SaveMode，它指定如果存在的话如何处理现有数据。
重要的是要认识到这些保存模式不使用任何锁定，也不是原子的。
此外，当执行覆盖时，数据将在写入新数据之前被删除。
<1> SaveMode.ErrorExists
如果目标位置已经存在数据，那么抛出一个异常
<2> SaveMode.Append
如果目标位置已经存在数据，那么将数据追加进去
<3> SaveMode.Overwrite
如果目标位置已经存在数据，那么就将已经存在的数据删除，用新数据进行覆盖
<4> SaveMode.Ignore
如果目标位置已经存在数据，那么就忽略，不做任何操作
3.实际案例
我们在构建数仓时，通常会有几种模式，其中全量模式通常是每天删除全部旧数据，接着再插入新数据。
一般来说
df.write
.format("jdbc")
.mode(SaveMode.Overwrite)
.option("url", conf.url)
.option("driver", conf.sinkVendor.getDriverClassName)
.option("dbtable", s"${conf.sinkDB}.${conf.sinkTable}")
.option("user", conf.username)
.option("password", conf.password)
.option("truncate", true)
.save()
这里配置了option("truncate"，ture)
在使用SaveMode.Overwrite时，一般是先DELETE来删除数据，这里的删除是一条条删除，而配置了truncate之后，那么是整体清除。
问题：在truncate清除数据之后，新数据还没有插入进来，那么这时候查询数据会出现问题，即查询不到数据
优化：对于一些旧数据较少的表，可以进行事务性删除添加操作
val confB = spark.sparkContext.broadcast((conf.url, conf.username, conf.password, conf.sinkDB, conf.sinkTable))
df.repartition(1).foreachPartition(iterator => {
val (url, username, password, sinkDB, sinkTable) = confB.value
JDBCUtil.withAutoClose(url, username, password, new WithConnection {
override def withConnection(conn: Connection): Unit = {
val sinkColumns = JDBCUtil.getColumns(conn, sinkDB, sinkTable).toSeq
JDBCUtil.withTransaction(conn, new WithConnection {
override def withConnection(conn: Connection): Unit = {
/* *****
* 删除
* *****/
val del = conn.createStatement()
del.execute(s"DELETE FROM $sinkDB.$sinkTable")
/* *********
* 批量插入
* *********/
var batchInsertStat: PreparedStatement = null
while (iterator.hasNext) { // 主循环，遍历分区
val row = iterator.next()
val inserts = new mutable.ListBuffer[(Int, String, Int, Any)] // 索引、字段名、类型、值
for (idx <- row.schema.indices) {
val field = row.schema(idx)
breakable {
for (c <- sinkColumns) {
if (c.getName.equalsIgnoreCase(field.name)) {
val i = (idx, c.getName, c.getType.toInt, row.get(idx))
inserts += i
break
}
}
}
}
if (batchInsertStat == null) {
val sql = s"""
|INSERT INTO $sinkDB.$sinkTable
|(${inserts.map(_._2).mkString(",")})
|VALUES
|(${Array.fill(inserts.size)("?").mkString(",")})
""".stripMargin
batchInsertStat = conn.prepareStatement(sql)
}
for (i <- inserts.indices) {
val insert = inserts(i)
batchInsertStat.setObject(i + 1, insert._4, insert._3)
}
batchInsertStat.addBatch()
}
batchInsertStat.executeBatch()
}
})
}
})
})
网友评论