您好, 欢迎来到 !    登录 | 注册 | | 设为首页 | 收藏本站

streaming 存储到mysql

bubuko 2022/1/25 20:10:19 mysql 字数 1508 阅读 1558 来源 http://www.bubuko.com/infolist-5-1.html

代码入下: ~~~ java //计算生成Dstream val result = stream.map(x = (x.key(), x.value())).reduceByKey((x, y) = (x + y)) //将计算结果保存到mysql中 采用druib 连接池的方式 var selec ...

代码入下:

//计算生成Dstream
 val result = stream.map(x => (x.key(), x.value())).reduceByKey((x, y) => (x + y))

//将计算结果保存到mysql中 采用druib 连接池的方式
var selectResule: ResultSet = null
var insertResult: ResultSet = null
var updateResult: ResultSet = null
result.foreachRDD(rdd => {
rdd.foreachPartition(partitionOfRecords => {
var connect: Connection = null
//此处判断是否存在RDD分区不存在就不需要获取连接类否则会导致连接数量过多而报错
if (partitionOfRecords.hasNext) {
connect = new DruidUtil().GetDruidConnect//从连接池中获取连接
partitionOfRecords.foreach(pair => {
val selectSql = "select * from wordcount where word='" + pair._1 + "'";
selectResule = connect.createStatement().executeQuery(selectSql)
//判断数据库中是否含有相应额key值如果存在就更新数据不存在就插入数据(实际项目中这里一般采用readis或者hbase的存储方式)
if (!selectResule.next()) {
val sql = "insert into wordcount(word,wordcount) values('" + pair._1 + "','" + pair._2 + "')"
connect.prepareStatement(sql).execute()
} else {
val updateSql = "update wordcount set wordcount='" + pair._2 + "' where word='" + pair._1 + "'"
connect.prepareStatement(updateSql).execute()
}
})
//关闭连接
connect.close()
}
})
})

streaming 存储到mysql

原文:https://www.cnblogs.com/HarSenZhao/p/12360617.html


如果您也喜欢它,动动您的小指点个赞吧

除非注明,文章均由 laddyq.com 整理发布,欢迎转载。

转载请注明:
链接:http://laddyq.com
来源:laddyq.com
著作权归作者所有。商业转载请联系作者获得授权,非商业转载请注明出处。


联系我
置顶