代码入下：
//计算生成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()
}
})
})原文：https://www.cnblogs.com/HarSenZhao/p/12360617.html
如果您也喜欢它,动动您的小指点个赞吧
602392714
清零编程群