Spark将计算结果写⼊到Mysql中
今天主要来谈谈如何将Spark计算的结果写⼊到Mysql或者其他的关系型数据库⾥⾯。其实⽅式也很简单，代码如下：
package scala
import java.sql.{DriverManager, PreparedStatement, Connection}
import org.apache.spark.{SparkContext, SparkConf}
object RDDtoMysql {
case class Blog(name: String, count: Int)
def myFun(iterator: Iterator[(String, Int)]): Unit = {
var conn: Connection = null
var ps: PreparedStatement = null
val sql = "insert into blog(name, count) values (?, ?)"
try {
conn = Connection("jdbc:mysql://localhost:3306/spark",
    "root", "123456")
iterator.foreach(data => {
ps = conn.prepareStatement(sql)
ps.setString(1, data._1)
ps.setInt(2, data._2)
}
)
} catch {
case e: Exception => println("Mysql Exception")
} finally {
if (ps != null) {
ps.close()
}
if (conn != null) {
conn.close()
}
}
}
def main(args: Array[String]) {
val conf = new SparkConf().setAppName("RDDToMysql").setMaster("local")
val sc = new SparkContext(conf)
val data = sc.parallelize(List(("www", 10), ("iteblog", 20), ("com", 30)))
data.foreachPartition(myFun)
}
}
其实是通过foreachPartition遍历RDD的每个分区，并调⽤普通的Scala⽅法来写数据库。在运⾏程序之前需要确保数据库⾥⾯存在blog表，可以通过下⾯语句创建：
CREATE TABLE `blog` (
`name` varchar(255) NOT NULL,
`count` int(10) unsigned DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf-8
然后直接运⾏上述的代码即可。运⾏完成你就可以在数据库⾥⾯查询结果：
SELECT * FROM blog b;
www  10
iteblog  20mysql下载app
com  30
需要注意的是：
  1、你最好使⽤foreachPartition 函数来遍历RDD，并且在每台Work上⾯创建数据库的connection。
  2、如果你的数据库并发受限，可以通过控制数据的分区来减少并发。
  3、在插⼊Mysql的时候最好使⽤批量插⼊。
  4、确保你写⼊数据库过程能够处理失败，因为你插⼊数据库的过程可能会经过⽹络，这可能会导致数据插⼊到数据库失败。
  5、不建议将你的RDD数据写⼊到Mysql等关系型数据库中。

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。