在Spark Streaming的应用程序中，有时候需要将计算结果保存到数据库中，为了高效这里使用批量插入，结合c3po连接池，说明一下使用方法。
- 数据计算完成后，在foreachRDD中批量插入数据，因为是针对每一个partiton的数据操作，所以使用 rdd.foreachPartition，这里是一个批量插入页面PV和UV的操作，代码如下
//RDD[(String,Int,Int)] 的意思是RDD[(页面名称,UV,PV)]
data.foreachRDD((rdd:RDD[(String,Int,Int)],time:Time)=>{
rdd.foreachPartition(data=>{
//从连接池中获取一个连接
val conn = MDBManager.getMDBManager(isLocal).getConnection
conn.setAutoCommit(false)
val sql = "insert into tableName set pageName=?,uvNum=?,pvNum=?"
val preparedStatement = conn.prepareStatement(sql)
data.foreach(r => {
preparedStatement.setObject(1, r._1)
preparedStatement.setObject(2, r._2)
preparedStatement.setObject(3, r._3)
preparedStatement.addBatch()
})
//批量提交，如果数据量大，这里可以分批提交
preparedStatement.executeBatch()
conn.commit()
conn.close()
})
- 这里创建一个单例的MDBManager，并使用c3p0获取连接，代码如下
class MDBManager(isLocal:Boolean) extends Serializable{
private val cpds: ComboPooledDataSource = new ComboPooledDataSource(true);
private val prop = new Properties()
private var in:InputStream = _
isLocal match{
case true => in = getClass().getResourceAsStream("/c3p0.properties");
case false => in = new FileInputStream(new File(SparkFiles.get("c3p0.properties")))
}
try {
prop.load(in);
cpds.setJdbcUrl(prop.getProperty("jdbcUrl").toString());
cpds.setDriverClass(prop.getProperty("driverClass").toString());
cpds.setUser(prop.getProperty("user").toString());
cpds.setPassword(prop.getProperty("password").toString()); cpds.setMaxPoolSize(Integer.valueOf(prop.getProperty("maxPoolSize").toString())); cpds.setMinPoolSize(Integer.valueOf(prop.getProperty("minPoolSize").toString())); cpds.setAcquireIncrement(Integer.valueOf(prop.getProperty("acquireIncrement").toString())); cpds.setInitialPoolSize(Integer.valueOf(prop.getProperty("initialPoolSize").toString())); cpds.setMaxIdleTime(Integer.valueOf(prop.getProperty("maxIdleTime").toString()));
} catch {
case ex: Exception => ex.printStackTrace()
}
def getConnection:Connection={
try {
return cpds.getConnection();
} catch {
case ex:Exception => ex.printStackTrace()
null
}
}
}
object MDBManager{
var mdbManager:MDBManager=_
def getMDBManager(isLocal:Boolean):MDBManager={
synchronized{
if(mdbManager==null){
mdbManager = new MDBManager(isLocal)
}
}
mdbManager
}
}
因为本地模式和集群模式的不同获取c3p0.properties配置文件也不一样，代码中分别提供了两种获取配件文件的方式，通过参数isLocal来确定使用哪种方式。
- 由于使用的是mysql数据库和c3p0连接池，所以提交应用时需要添加mysql连接的jar包和c3p0的jar包，在Spark-submit中添加参数
--jars /usr/local/spark1.3/lib/mysql-connector-java-5.1.38-bin.jar,/usr/local/spark1.3/lib/c3p0-0.9.1.2.jar
- 提交应用时添加c3p0的配置文件,在Spark-submit中添加参数
--files /usr/local/spark1.3/conf/c3p0.properties
