在MySQL创建表

create table day_netType_access_topn_stat (
day varchar(8) not null,
uid bigint(10) not null,
times bigint(10) not null,
primary key (day, uid)
)

查看表结构：

创建Entity

package cn.ac.iie.log/*** 每天访问次数实体类* @param day* @param uid* @param times*/
case class DayNetTypeAccessStat (day: String, uid:Long, times:Long)

创建Dao

insert

package cn.ac.iie.logimport java.sql.{Connection, PreparedStatement}import scala.collection.mutable.ListBuffer/*** 各个维度统计的DAO操作*/
object StatDao {/*** 批量保存DayVideoAccessStat到数据库** @param list*/def insertNetTypeAccessTopN(list: ListBuffer[DayNetTypeAccessStat]): Unit = {var connection: Connection = nullvar pstmt: PreparedStatement = nulltry {connection = MysqlUtils.getConnection()// 设置手动提交connection.setAutoCommit(false)val sql = "insert into day_netType_access_topn_stat (day, uid, times) values (?,?,?)"pstmt = connection.prepareStatement(sql)for (ele <- list) {pstmt.setString(1, ele.day)pstmt.setLong(2, ele.uid)pstmt.setLong(3, ele.times)pstmt.addBatch()}pstmt.executeBatch() // 执行批量处理// 手动提交connection.commit()} catch {case e: Exception => e.printStackTrace()} finally {MysqlUtils.release(connection, pstmt)}}
}

这里insert数据时，最好使用批处理，提交使用batch操作，手动提交。

将数据保存到Mysql中

package cn.ac.iie.logimport org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._import scala.collection.mutable.ListBuffer/*** TopN 统计spark作业*/
object TopNStatJob {def main(args: Array[String]): Unit = {val spark = SparkSession.builder().appName("TopNStatJob").config("spark.sql.sources.partitionColumnTypeInference.enabled","false").master("local[2]").getOrCreate()val accessDF = spark.read.format("parquet").load("file:///E:/test/clean")
//    accessDF.printSchema()accessDF.show(false)// 最受欢迎的TopN netTypenetTypeAccessTopNStat(spark, accessDF)spark.stop}/*** 最受欢迎的TopN netType* @param spark* @param accessDF*/def netTypeAccessTopNStat(spark: SparkSession, accessDF: DataFrame): Unit = {accessDF.createOrReplaceTempView("access_logs")val wifiAccessTopNDF = spark.sql("select day,uid,count(1) as times from access_logs where day='20190702' and netType='wifi' group by day,uid order by times desc")
//    wifiAccessTopNDF.show(false)// 将统计结果写入到Mysql中try{wifiAccessTopNDF.foreachPartition(partitionOfRecords => {val list = new ListBuffer[DayNetTypeAccessStat]partitionOfRecords.foreach(info => {val day = info.getAs[String]("day")val uid = info.getAs[Long]("uid")val times = info.getAs[Long]("times")list.append(DayNetTypeAccessStat(day, uid, times))})StatDao.insertNetTypeAccessTopN(list)})} catch {case e: Exception => e.printStackTrace()}}
}

SparkSQL 将统计结果保存到Mysql相关推荐

SparkStreaming读取Kafka的Json数据然后保存到MySQL
一般我们使用SparkStreaming消费kafka数据,获取到数据后解析,使用JDBC的方式写入数据库,如下所示. 以上的方式没什么毛病,但是当我们消费的kafka数据类型比较多样的时候,我们需要 ...
kafka偏移量保存到mysql里_SparkStreaming+kafka保存offset的偏移量到mysql案例
MySQL创建存储offset的表格 mysql> use test mysql> create table hlw_offset( topic varchar(32), groupid ...
mysql打印语句_大数据挖掘—（八）：scrapy爬取数据保存到MySql数据库
(大数据挖掘-(七):读懂MySql数据库操作)(大数据挖掘神器--scrapy spider爬虫框架(五):解析多层网页) 通过往期的文章分享,我们了解了如何爬取想要的数据到Items中,也了解了如 ...
pandas对象保存到mysql出错提示“BLOB/TEXT column used in key specification without a key length”解决办法
问题将DataFrame数据保存到mysql中时,出现错误提示: BLOB/TEXT column used in key specification without a key length 原因 ...
android数据库给单选赋值,如何使用android studio将单选按钮的值保存到mysql数据库？...
我想创建一个投票应用程序,其中有不同的职位,其中有两个职位每个都有两个联系人.我希望当我选择一个人记录到数据库. [职位的disgn与它的参赛者部] 下面是XML代码的一部分:如何使用android ...
mysql t 保存_检查 (调试) - 离线消息保存到 MySQL - 《EMQ X Enterprise v4.1 中文文档》 - 书栈网 · BookStack...
离线消息保存到 MySQL 搭建 MySQL 数据库,并设置用户名密码为 root/public,以 MacOS X 为例: $ brew install mysql $ brew services ...
python股票接口_Python 从 sina 股票数据接口读取数据，并保存到 MySQL 数据库
说明从 sina 的数据接口获取数据,之后,保存到 MySql 数据库文件:getDataFromSina.py ''' Created on 2018年2月11日 @author: Livon ...
localdate存mysql相差一天_如何在保存到mySQL数据库时阻止LocalDate更改
使用JPA CriteriaBuilder API将LocalDate字段(例如'2017-09-27′)保存到mySQL Date列时,结果不同(例如'2017-09-26′). 我已经使用SELE ...
将labview连续数据保存到mysql数据库器
这一篇是在之前完成Labview和mysql连接,并且进行了简单的CRUD删除的基础上来的.我们一般不会拿Labview来做学生这种数据管理系统,而是对于基本传感器数据的采集和保存,而传感器采集数据会 ...

SparkSQL 将统计结果保存到Mysql

在MySQL创建表

创建Entity

创建Dao

insert

将数据保存到Mysql中

SparkSQL 将统计结果保存到Mysql相关推荐

最新文章

热门文章