区块链技术博客
www.b2bchain.cn

SparkStreaming项目实战,实时计算pv和uv(硬肝)求职学习资料

本文介绍了SparkStreaming项目实战,实时计算pv和uv(硬肝)求职学习资料,有助于帮助完成毕业设计以及求职,是一篇很好的资料。

对技术面试,学习经验等有一些体会,在此分享。

最近有个需求,实时统计pv,uv,结果按照date,hour,pv,uv来展示,按天统计,第二天重新统计,当然了实际还需要按照类型字段分类统计pv,uv,比如按照date,hour,pv,uv,type来展示。这里介绍最基本的pv,uv的展示。

id uv pv date hour
1 155599 306053 2018-07-27 18

关于什么是pv,uv,可以参见这篇博客:https://blog.csdn.net/petermsh/article/details/78652246

1、项目流程

SparkStreaming项目实战,实时计算pv和uv(硬肝)
日志数据从flume采集过来,落到hdfs供其它离线业务使用,也会sink到kafka,sparkStreaming从kafka拉数据过来,计算pv,uv,uv是用的redis的set集合去重,最后把结果写入mysql数据库,供前端展示使用。

2、具体过程

1)pv的计算

拉取数据有两种方式,基于received和direct方式,这里用direct直拉的方式,用的mapWithState算子保存状态,这个算子与updateStateByKey一样,并且性能更好。当然了实际中数据过来需要经过清洗,过滤,才能使用。

定义一个状态函数

// 实时流量状态更新函数   val mapFunction = (datehour:String, pv:Option[Long], state:State[Long]) => {     val accuSum = pv.getOrElse(0L) + state.getOption().getOrElse(0L)     val output = (datehour,accuSum)     state.update(accuSum)     output   }

这样就很容易的把pv计算出来了。

2)uv的计算

uv是要全天去重的,每次进来一个batch的数据,如果用原生的reduceByKey或者groupByKey对配置要求太高,在配置较低情况下,我们申请了一个93G的redis用来去重,原理是每进来一条数据,将date作为key,guid加入set集合,20秒刷新一次,也就是将set集合的尺寸取出来,更新一下数据库即可。

helper_data.foreachRDD(rdd => {         rdd.foreachPartition(eachPartition => {         // 获取redis连接           val jedis = getJedis           eachPartition.foreach(x => {             // 省略若干...             jedis.sadd(key,x._2)             // 设置存储每天的数据的set过期时间,防止超过redis容量,这样每天的set集合,定期会被自动删除             jedis.expire(key,ConfigFactory.rediskeyexists)           })           // 关闭连接           closeJedis(jedis)         })       })

3)结果保存到数据库

结果保存到mysql,数据库,10秒刷新一次数据库,前端展示刷新一次,就会重新查询一次数据库,做到实时统计展示pv,uv的目的。
“`java
/**
* 插入数据
* @param data (addTab(datehour)+helperversion)
* @param tbName
* @param colNames
/ def insertHelper(data: DStream[(String, Long)], tbName: String, colNames: String): Unit = {
data.foreachRDD(rdd => {
val tmp_rdd = rdd.map(x => x._1.substring(11, 13).toInt)
if (!rdd.isEmpty()) {
val hour_now = tmp_rdd.max() // 获取当前结果中最大的时间,在数据恢复中可以起作用
rdd.foreachPartition(eachPartition => {
try {
val jedis = getJedis
val conn = MysqlPoolUtil.getConnection()
conn.setAutoCommit(false)
val stmt = conn.createStatement()
eachPartition.foreach(x => {
// val sql = ….
// 省略若干
stmt.addBatch(sql)
})
closeJedis(jedis)
stmt.executeBatch() // 批量执行sql语句
conn.commit()
conn.close()
} catch {
case e: Exception => {
logger.error(e)
logger2.error(HelperHandle.getClass.getSimpleName + e)
}
}
})
}
})
}

// 计算当前时间距离次日零点的时长(毫秒)
def resetTime = {
val now = new Date()
val todayEnd = Calendar.getInstance
todayEnd.set(Calendar.HOUR_OF_DAY, 23) // Calendar.HOUR 12小时制
todayEnd.set(Calendar.MINUTE, 59)
todayEnd.set(Calendar.SECOND, 59)
todayEnd.set(Calendar.MILLISECOND, 999)
todayEnd.getTimeInMillis – now.getTime
}

最近有个需求,实时统计pv,uv,结果按照date,hour,pv,uv来展示,按天统计,第二天重新统计,当然了实际还需要按照类型字段分类统计pv,uv,比如按照date,hour,pv,uv,type来展示。这里介绍最基本的pv,uv的展示。

id uv pv date hour
1 155599 306053 2018-07-27 18

关于什么是pv,uv,可以参见这篇博客:https://blog.csdn.net/petermsh/article/details/78652246

1、项目流程

SparkStreaming项目实战,实时计算pv和uv(硬肝)
日志数据从flume采集过来,落到hdfs供其它离线业务使用,也会sink到kafka,sparkStreaming从kafka拉数据过来,计算pv,uv,uv是用的redis的set集合去重,最后把结果写入mysql数据库,供前端展示使用。

2、具体过程

1)pv的计算

拉取数据有两种方式,基于received和direct方式,这里用direct直拉的方式,用的mapWithState算子保存状态,这个算子与updateStateByKey一样,并且性能更好。当然了实际中数据过来需要经过清洗,过滤,才能使用。

定义一个状态函数

// 实时流量状态更新函数   val mapFunction = (datehour:String, pv:Option[Long], state:State[Long]) => {     val accuSum = pv.getOrElse(0L) + state.getOption().getOrElse(0L)     val output = (datehour,accuSum)     state.update(accuSum)     output   }

这样就很容易的把pv计算出来了。

2)uv的计算

uv是要全天去重的,每次进来一个batch的数据,如果用原生的reduceByKey或者groupByKey对配置要求太高,在配置较低情况下,我们申请了一个93G的redis用来去重,原理是每进来一条数据,将date作为key,guid加入set集合,20秒刷新一次,也就是将set集合的尺寸取出来,更新一下数据库即可。

helper_data.foreachRDD(rdd => {         rdd.foreachPartition(eachPartition => {         // 获取redis连接           val jedis = getJedis           eachPartition.foreach(x => {             // 省略若干...             jedis.sadd(key,x._2)             // 设置存储每天的数据的set过期时间,防止超过redis容量,这样每天的set集合,定期会被自动删除             jedis.expire(key,ConfigFactory.rediskeyexists)           })           // 关闭连接           closeJedis(jedis)         })       })

3)结果保存到数据库

结果保存到mysql,数据库,10秒刷新一次数据库,前端展示刷新一次,就会重新查询一次数据库,做到实时统计展示pv,uv的目的。
“`java
/**
* 插入数据
* @param data (addTab(datehour)+helperversion)
* @param tbName
* @param colNames
/ def insertHelper(data: DStream[(String, Long)], tbName: String, colNames: String): Unit = {
data.foreachRDD(rdd => {
val tmp_rdd = rdd.map(x => x._1.substring(11, 13).toInt)
if (!rdd.isEmpty()) {
val hour_now = tmp_rdd.max() // 获取当前结果中最大的时间,在数据恢复中可以起作用
rdd.foreachPartition(eachPartition => {
try {
val jedis = getJedis
val conn = MysqlPoolUtil.getConnection()
conn.setAutoCommit(false)
val stmt = conn.createStatement()
eachPartition.foreach(x => {
// val sql = ….
// 省略若干
stmt.addBatch(sql)
})
closeJedis(jedis)
stmt.executeBatch() // 批量执行sql语句
conn.commit()
conn.close()
} catch {
case e: Exception => {
logger.error(e)
logger2.error(HelperHandle.getClass.getSimpleName + e)
}
}
})
}
})
}

// 计算当前时间距离次日零点的时长(毫秒)
def resetTime = {
val now = new Date()
val todayEnd = Calendar.getInstance
todayEnd.set(Calendar.HOUR_OF_DAY, 23) // Calendar.HOUR 12小时制
todayEnd.set(Calendar.MINUTE, 59)
todayEnd.set(Calendar.SECOND, 59)
todayEnd.set(Calendar.MILLISECOND, 999)
todayEnd.getTimeInMillis – now.getTime
}

最近有个需求,实时统计pv,uv,结果按照date,hour,pv,uv来展示,按天统计,第二天重新统计,当然了实际还需要按照类型字段分类统计pv,uv,比如按照date,hour,pv,uv,type来展示。这里介绍最基本的pv,uv的展示。

id uv pv date hour
1 155599 306053 2018-07-27 18

关于什么是pv,uv,可以参见这篇博客:https://blog.csdn.net/petermsh/article/details/78652246

1、项目流程

SparkStreaming项目实战,实时计算pv和uv(硬肝)
日志数据从flume采集过来,落到hdfs供其它离线业务使用,也会sink到kafka,sparkStreaming从kafka拉数据过来,计算pv,uv,uv是用的redis的set集合去重,最后把结果写入mysql数据库,供前端展示使用。

2、具体过程

1)pv的计算

拉取数据有两种方式,基于received和direct方式,这里用direct直拉的方式,用的mapWithState算子保存状态,这个算子与updateStateByKey一样,并且性能更好。当然了实际中数据过来需要经过清洗,过滤,才能使用。

定义一个状态函数

// 实时流量状态更新函数   val mapFunction = (datehour:String, pv:Option[Long], state:State[Long]) => {     val accuSum = pv.getOrElse(0L) + state.getOption().getOrElse(0L)     val output = (datehour,accuSum)     state.update(accuSum)     output   }

这样就很容易的把pv计算出来了。

2)uv的计算

uv是要全天去重的,每次进来一个batch的数据,如果用原生的reduceByKey或者groupByKey对配置要求太高,在配置较低情况下,我们申请了一个93G的redis用来去重,原理是每进来一条数据,将date作为key,guid加入set集合,20秒刷新一次,也就是将set集合的尺寸取出来,更新一下数据库即可。

helper_data.foreachRDD(rdd => {         rdd.foreachPartition(eachPartition => {         // 获取redis连接           val jedis = getJedis           eachPartition.foreach(x => {             // 省略若干...             jedis.sadd(key,x._2)             // 设置存储每天的数据的set过期时间,防止超过redis容量,这样每天的set集合,定期会被自动删除             jedis.expire(key,ConfigFactory.rediskeyexists)           })           // 关闭连接           closeJedis(jedis)         })       })

3)结果保存到数据库

结果保存到mysql,数据库,10秒刷新一次数据库,前端展示刷新一次,就会重新查询一次数据库,做到实时统计展示pv,uv的目的。
“`java
/**
* 插入数据
* @param data (addTab(datehour)+helperversion)
* @param tbName
* @param colNames
/ def insertHelper(data: DStream[(String, Long)], tbName: String, colNames: String): Unit = {
data.foreachRDD(rdd => {
val tmp_rdd = rdd.map(x => x._1.substring(11, 13).toInt)
if (!rdd.isEmpty()) {
val hour_now = tmp_rdd.max() // 获取当前结果中最大的时间,在数据恢复中可以起作用
rdd.foreachPartition(eachPartition => {
try {
val jedis = getJedis
val conn = MysqlPoolUtil.getConnection()
conn.setAutoCommit(false)
val stmt = conn.createStatement()
eachPartition.foreach(x => {
// val sql = ….
// 省略若干
stmt.addBatch(sql)
})
closeJedis(jedis)
stmt.executeBatch() // 批量执行sql语句
conn.commit()
conn.close()
} catch {
case e: Exception => {
logger.error(e)
logger2.error(HelperHandle.getClass.getSimpleName + e)
}
}
})
}
})
}

// 计算当前时间距离次日零点的时长(毫秒)
def resetTime = {
val now = new Date()
val todayEnd = Calendar.getInstance
todayEnd.set(Calendar.HOUR_OF_DAY, 23) // Calendar.HOUR 12小时制
todayEnd.set(Calendar.MINUTE, 59)
todayEnd.set(Calendar.SECOND, 59)
todayEnd.set(Calendar.MILLISECOND, 999)
todayEnd.getTimeInMillis – now.getTime
}

部分转自互联网,侵权删除联系

赞(0) 打赏
部分文章转自网络,侵权联系删除b2bchain区块链学习技术社区 » SparkStreaming项目实战,实时计算pv和uv(硬肝)求职学习资料
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

b2b链

联系我们联系我们