spark - 将RDD保存到RMDB(MYSQL)数据库中
2016-01-28 13:07
573 查看
SCALA连接数据库批量插入:
scala> import java.sql.DriverManager
scala> var url = "jdbc:mysql://localhost:3306/mydb?useUnicode=true&characterEncoding=utf8"
scala> var username = "cui"
scala> var password = "dbtest"
scala> val conn= DriverManager.getConnection(url,username,password)
scala> val pstat = conn.prepareStatement ("INSERT INTO `TEST` (`ID`, `AGE`) VALUES (?, ?)")
scala> pstat.clearBatch
scala> pstat.setInt(1,501)
scala> pstat.setInt(2,501)
scala> pstat.addBatch
scala> pstat.setInt(1,502)
scala> pstat.setInt(2,502)
scala> pstat.addBatch
scala> pstat.setInt(1,503)
scala> pstat.setInt(2,503)
scala> pstat.addBatch
scala> pstat.executeBatch
res24: Array[Int] = Array(1, 1, 1)
RDD保存到数据库:
Just use foreachPartition to create and execute a SQL statement via JDBC over a batch of records. The code is just normal JDBC code.
https://community.cloudera.com/t5/Advanced-Analytics-Apache-Spark/Spark-Streaming-save-output-to-mysql-DB/td-p/25607
scala> import java.sql.DriverManager
scala> var url = "jdbc:mysql://localhost:3306/mydb?useUnicode=true&characterEncoding=utf8"
scala> var username = "cui"
scala> var password = "dbtest"
scala> val conn= DriverManager.getConnection(url,username,password)
scala> val pstat = conn.prepareStatement ("INSERT INTO `TEST` (`ID`, `AGE`) VALUES (?, ?)")
scala> pstat.clearBatch
scala> pstat.setInt(1,501)
scala> pstat.setInt(2,501)
scala> pstat.addBatch
scala> pstat.setInt(1,502)
scala> pstat.setInt(2,502)
scala> pstat.addBatch
scala> pstat.setInt(1,503)
scala> pstat.setInt(2,503)
scala> pstat.addBatch
scala> pstat.executeBatch
res24: Array[Int] = Array(1, 1, 1)
RDD保存到数据库:
Just use foreachPartition to create and execute a SQL statement via JDBC over a batch of records. The code is just normal JDBC code.
https://community.cloudera.com/t5/Advanced-Analytics-Apache-Spark/Spark-Streaming-save-output-to-mysql-DB/td-p/25607
import java.sql.DriverManager var data = sc.parallelize(Array( (1,10) ,(1,100), (1,1000), (1,10000),(2,10) ,(2,100), (2,1000), (2,10000) ),2 ) data.foreachPartition( it =>{ var url = "jdbc:mysql://localhost:3306/mydb?useUnicode=true&characterEncoding=utf8" val conn= DriverManager.getConnection(url,"username","password") val pstat = conn.prepareStatement ("INSERT INTO `TEST` (`ID`, `AGE`) VALUES (?, ?)") for (obj <-it){ pstat.setInt(1,obj._1) pstat.setInt(2,obj._2) pstat.addBatch } try{ pstat.executeBatch }finally{ pstat.close conn.close } } )
相关文章推荐
- 设计模式 - 单例模式mysql数据库操作类
- MySQL中KEY与INDEX的区别
- mysql读写分离(使用Atlas实现)
- MySQL中EXPLAIN详解
- MySql 账户管理
- mysql聚集索引的优缺点
- [转]向facebook学习,通过协程实现mysql查询的异步化
- mysql 命令行执行sql语句
- unix下mysql丢失或者遗忘密码怎么办
- SqlServer实时数据同步到MySql
- MySQL 插入时间
- mysql : Lock wait timeout exceeded; try restarting transaction
- mysql 的 几种 timeout(更新中......)
- SqlServer实时数据同步到mysql
- 破解mysql root密码的几种方法
- mysqldump
- MySQL日期时间函数大全
- pdo_mysql安装
- MySQL开发规范与使用技巧总结
- mysql 添加索引 mysql 如何创建索引