你好,游客 登录
背景:
阅读新闻

Spark Streaming 中使用c3p0连接池操作mysql数据库

[日期:2016-06-20] 来源:简书  作者:_海纳百川 [字体: ]

在Spark Streaming的应用程序中,有时候需要将计算结果保存到数据库中,为了高效这里使用批量插入,结合c3po连接池,说明一下使用方法。

  1. 数据计算完成后,在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()
    })
  2. 这里创建一个单例的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来确定使用哪种方式。

  3. 由于使用的是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
  4. 提交应用时添加c3p0的配置文件,在Spark-submit中添加参数
    --files /usr/local/spark1.3/conf/c3p0.properties




收藏 推荐 打印 | 录入: | 阅读:
本文评论   查看全部评论 (0)
表情: 表情 姓名: 字数
点评:
       
评论声明
  • 尊重网上道德,遵守中华人民共和国的各项有关法律法规
  • 承担一切因您的行为而直接或间接导致的民事或刑事法律责任
  • 本站管理人员有权保留或删除其管辖留言中的任意内容
  • 本站有权在网站内转载或引用您的评论
  • 参与本评论即表明您已经阅读并接受上述条款