/**
* User: 过往记忆
* Date: 15-03-10
* Time: 上午07:30
* bolg:
* 本文地址：/archives/1275
* 过往记忆博客，专注于hadoop、hive、spark、shark、flume的技术博客，大量的干货
* 过往记忆博客微信公共帐号：iteblog_hadoop
*/
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 = 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中】（https://www.iteblog.com/archives/1275.html）
楼主, 你好, 请问sparkstreaming可以整合springboot线程池和druid连接池吗?可以实现每个分区简历连接druid连接池吗?
先每个分区实现实现循环遍历,通过线程池对记录的key加锁,保证相同记录的数据同一条只有一个能够进入,然后每条进入的记录获得连接池的一个连接插入记录?
楼主, 你好, 请问sparkstreaming可以整合springboot线程池和druid连接池吗?可以实现每个分区简历连接druid连接池吗? 先每个分区实现
从RDD到Mysql这个过程产生数据的重复插入该怎么解决呢
你给每条记录设置一个主键，下一次再插入的时候就会出现异常，你处理一下就行。
请问如果我要是都HDFS上的文件 该怎么写这个data 写来写去老是说不符合myFun函数