Spark将计算结果写入到Mysql中

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 = DriverManager.getConnection("jdbc:mysql://localhost:3306/spark", 　　　　"root", "123456") iterator.foreach(data => { ps = conn.prepareStatement(sql) ps.setString(1, data._1) ps.setInt(2, data._2) ps.executeUpdate() } ) } 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　　20
com　　30

需要注意的是：
　　1、你最好使用foreachPartition 函数来遍历RDD，并且在每台Work上面创建数据库的connection。
　　2、如果你的数据库并发受限，可以通过控制数据的分区来减少并发。
　　3、在插入Mysql的时候最好使用批量插入。
　　4、确保你写入数据库过程能够处理失败，因为你插入数据库的过程可能会经过网络，这可能会导致数据插入到数据库失败。
　　5、不建议将你的RDD数据写入到Mysql等关系型数据库中。

Spark将计算结果写入到Mysql中

标签：localhost   方法   case   create   code   log   批量插入   rman   sign