import java.sql.{Connection, ResultSet} import com.jolbox.bonecp.{BoneCP, BoneCPConfig} import org.slf4j.LoggerFactory object ConnectionPool { val logger = LoggerFactory.getLogger(this.getClass) private val connectionPool = { try{ Class.forName("com.mysql.jdbc.Driver") val config = new BoneCPConfig() config.setJdbcUrl("jdbc:mysql://192.168.0.46:3306/test") config.setUsername("test") config.setPassword("test") config.setMinConnectionsPerPartition(2) config.setMaxConnectionsPerPartition(5) config.setPartitionCount(3) config.setCloseConnectionWatch(true) config.setLogStatementsEnabled(true) Some(new BoneCP(config)) } catch { case exception:Exception=> logger.warn("Error in creation of connection pool"+exception.printStackTrace()) None } } def getConnection:Option[Connection] ={ connectionPool match { case Some(connPool) => Some(connPool.getConnection) case None => None } } def closeConnection(connection:Connection): Unit = { if(!connection.isClosed) connection.close() } }
import java.sql.{Connection, DriverManager, PreparedStatement} import org.apache.spark.streaming.kafka.KafkaUtils import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.{SparkConf, SparkContext} import org.slf4j.LoggerFactory /** * 记录最近五秒钟的数据 */ object RealtimeCount1{ case class Loging(vtime:Long,muid:String,uid:String,ucp:String,category:String,autoSid:Int,dealerId:String,tuanId:String,newsId:String) case class Record(vtime:Long,muid:String,uid:String,item:String,types:String) val logger = LoggerFactory.getLogger(this.getClass) def main(args: Array[String]) { val argc = new Array[String](4) argc(0) = "10.0.0.37" argc(1) = "test-1" argc(2) = "test22" argc(3) = "1" val Array(zkQuorum, group, topics, numThreads) = argc val sparkConf = new SparkConf().setAppName("RealtimeCount").setMaster("local[2]") val sc = new SparkContext(sparkConf) val ssc = new StreamingContext(sc, Seconds(5)) val topicMap = topics.split(",").map((_,numThreads.toInt)).toMap val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(x=>x._2) val sql = "insert into loging_realtime1(vtime,muid,uid,item,category) values (?,?,?,?,?)" val tmpdf = lines.map(_.split("\t")).map(x=>Loging(x(9).toLong,x(1),x(0),x(3),x(25),x(18).toInt,x(29),x(30),x(28))).filter(x=>(x.muid!=null && !x.muid.equals("null") && !("").equals(x.muid))).map(x=>Record(x.vtime,x.muid,x.uid,getItem(x.category,x.ucp,x.newsId,x.autoSid.toInt,x.dealerId,x.tuanId),getType(x.category,x.ucp,x.newsId,x.autoSid.toInt,x.dealerId,x.tuanId))) tmpdf.filter(x=>x.types!=null).foreachRDD{rdd => //rdd.foreach(println) rdd.foreachPartition(partitionRecords=>{ val connection = ConnectionPool.getConnection.getOrElse(null) if(connection!=null){ partitionRecords.foreach(record=>process(connection,sql,record)) ConnectionPool.closeConnection(connection) } }) } ssc.start() ssc.awaitTermination() } def getItem(category:String,ucp:String,newsId:String,autoSid:Int,dealerId:String,tuanId:String):String = { if(category!=null && !category.equals("null")){ val pattern = "http://www.ihaha.com/\\d{4}-\\d{2}-\\d{2}/\\d{9}.html" val matcher = ucp.matches(pattern) if(matcher) { ucp.substring(33,42) }else{ null } }else if(autoSid!=0){ autoSid.toString }else if(dealerId!=null && !dealerId.equals("null")){ dealerId }else if(tuanId!=null && !tuanId.equals("null")){ tuanId }else{ null } } def getType(category:String,ucp:String,newsId:String,autoSid:Int,dealerId:String,tuanId:String):String = { if(category!=null && !category.equals("null")){ val pattern = "100000726;100000730;\\d{9};\\d{9}" val matcher = category.matches(pattern) val pattern1 = "http://www.chexun.com/\\d{4}-\\d{2}-\\d{2}/\\d{9}.html" val matcher1 = ucp.matches(pattern1) if(matcher1 && matcher) { "nv" }else if(newsId!=null && !newsId.equals("null") && matcher1){ "ns" }else if(matcher1){ "ne" }else{ null } }else if(autoSid!=0){ "as" }else if(dealerId!=null && !dealerId.equals("null")){ "di" }else if(tuanId!=null && !tuanId.equals("null")){ "ti" }else{ null } } def process(conn:Connection,sql:String,data:Record): Unit ={ try{ val ps : PreparedStatement = conn.prepareStatement(sql) ps.setLong(1,data.vtime) ps.setString(2,data.muid) ps.setString(3,data.uid) ps.setString(4,data.item) ps.setString(5,data.types) ps.executeUpdate() }catch{ case exception:Exception=> logger.warn("Error in execution of query"+exception.printStackTrace()) } } }
使用连接池的方式获取connection
相关推荐
(1)利用SparkStreaming从文件目录读入日志信息,日志内容包含: ”日志级别、函数名、日志内容“ 三个字段,字段之间以空格拆分。请看数据源的文件。 (2)对读入都日志信息流进行指定筛选出日志级别为error或warn...
通过该示例,您将了解到如何使用Spark Streaming和Kafka处理实时数据流,以及如何将处理后的数据保存到MySQL数据库中。示例涵盖了从环境搭建到代码实现的全过程,帮助您快速上手实时数据处理的开发。提供了一个完整...
7.外部数据源(HBase和MySQL) 8.广播变量和累加器 9.Spark 内核调度 10.Spark 并行度 第三章、SparkSQL 模块 1.快速入门:词频统计 2.SparkSQL 概述 3.DataFrame 4.RDD与DataFrame转换 5.数据分析SQL和DSL 6.案例:...
该项目是一个支持HBase/MySQL Binlog的Spark structured streaming数据源。通过该项目,开发者可以学习并实践Spark和Scala技术的应用,为后续的大数据处理奠定基础。系统界面友好,易于操作,适合用于各类大数据处理...
3.使用Spark Streaming对数据进行清洗、加工、处理,形成最终要展示的指标,存入MySQL,提供给前端开发。 4.研究数据结果,剖析有效信息,提出指导性意见与结论(书面、口头等)。 辅助安装小型气象站与自动化监测...
本项目基于搜狗实验室的用户查询日志数据,对数据集的简单处理,将数据集上传到集群上,编写并使用日志模拟数据产生的程序和脚本作为数据源,将01作为聚合节点,利用flume将数据下沉到Kafka消息队列中;接着通过...
优雅的交互方式,支持多种datasource/sink,多数据源混算 spark常驻服务,基于zookeeper的引擎自动发现 负载均衡,多个引擎随机执行 多session模式实现并行查询 采用spark的FAIR调度,避免资源被大任务独占 基于...
浅谈数据仓库和⼤数据 浅谈数据仓库和⼤...互联⽹⾏业 离线仓库架构:Sqoop+hadoop+hive/spark+mysql/hbase+echarts/tableau/highchars 实时架构:flume+kafka+storm/spark streaming+hbase/redis+echarts/tableau/highc
sqoop 数据迁移⼯具,⽤来在不同数据存储软件之间进⾏数据传输的开源软件 DataX 阿⾥巴巴开源的离线数据同步⼯具,⽤于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间...
⼤数据应⽤测试经验总结 ⼤数据应⽤测试经验总结 ⼤数据应⽤测试过程与传统的web系统有较⼤的不同,⼤数据应⽤测试通常会分为web侧和ETL侧测试,web侧基本就是功能测试,⽽ETL(Extracting-Transfroming- Loading)...
⼤数据的基础知识 1、⼤数据的基础知识 ⼤数据的概念 ⼤数据(big data),IT⾏业术语,是指⽆法在⼀定时间范围内⽤常规软件⼯具进⾏捕捉、管理和处理的数据集合,是需要新处理模式才能 具有更强的决策⼒、洞察发现...