Spark项目实战-卡口流量统计

2023-10-25 14:01
文章标签 实战 统计 项目 流量 spark 卡口

本文主要是介绍Spark项目实战-卡口流量统计，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

一、卡口介绍

在这里插入图片描述
卡口摄像头正对车道安装，拍摄正面照片。
功能：抓拍正面特征

这种摄像头多安装在国道、省道、高速公路的路段上、或者城区和郊区交接的主要路口，用来抓拍超速、进出城区车辆等行为。它进行的是车辆正面抓拍，可以清晰地看到驾驶员及前台乘客的面容及行为。有一些则是专门摄像车的尾部，所以当车开过此类测速摄像头后不要马上提速，建议至少要跑出500米后再提速。这就是有人认为的没有超速为什么也照样被拍的原因。此类摄像头应该是集成照明设备。

在这里插入图片描述

在这里插入图片描述

卡口：三车道、三个摄像头共同组成卡口A

二、表介绍

monitor_flow_action

日期卡口ID摄像头编号车牌号拍摄时间车速道路ID区域ID
datemonitor_idcamera_idcaraction_timespeedroad_idarea_id
2018-11-05000533745京C601592018-11-05 20:43:471983604

monitor_camera_info

卡扣号摄像头编号
monitor_idcamera_id
000680522
000629268

areaId2AreaInfoRDD

area_idarea_name
区域ID区域Name

tmp_car_flow_basic
= areaId2AreaInfoRDD + monitor_flow_action

卡口ID车牌号道路ID区域ID区域Name
monitor_idcarroad_idarea_idarea_name
0005京C601593604

tmp_area_road_flow_count

area_nameroad_idcar_countmonitor_infos
区域ID道路ID车count详情
0436500006=20|0002=30

areaTop3Road

area_nameroad_idcar_countmonitor_infosflow_level
区域ID道路ID车count详情流量等级
0436500006=20|0002=30D

三、分析需求

3.1 卡口正常数、异常数

在这里插入图片描述

统计异常摄像头的思路：流量表 rightJoin 摄像头表，过滤流量表为空的

//------------------------------统计卡口摄像头通过的车辆的合计----------------------------
//| 2023-10-24 | 0005 | 33745 | 京C60159 | 2018-11-05 20:43:47 | 198 | 36 | 04 |
val flowDF:DataFrame =  sparkSession.sql("select * from monitor_flow_action where data = '2023-10-24' ")//((0005:33745),1)
val mcRdd: RDD[(String, Int)] = flowDF.map(e => Tuple2((e.getString(1) + ":" + e.getString(2)),1)).rdd//((0005:33745),99)
val flowRdd: RDD[(String, Int)] = mcRdd.reduceByKey(_+_)//------------------------------统计卡口所有的摄像头----------------------------
//| 0006 | 29268 |
val cameraDF:DataFrame = sparkSession.sql("select * from monitor_camera_info")//((0006,29268),1)
val cameraRdd: RDD[(String, Int)] = cameraDF.map(e => ((e.getString(0) + ":" + e.getString(1)),1)).rdd//------------------------------合并车流量和摄像头RDD----------------------------
val allRDD: RDD[(String, (Option[Int], Int))] = flowRdd.rightOuterJoin(cameraRdd).filter(e => e._2._1.isEmpty)

在这里插入图片描述

3.2 camera 正常数、异常数、详情

    //---------------------开始操作车流量信息，假设任务编号为1 日期参数为今天val flowInfo: RDD[(String, String)] = sparkSession.sql("select * from monitor_flow_action where date = '2021-08-23' ").rdd.map(row => (row.getString(1), row)).groupByKey().map(ele => {val monitorId: String = ele._1val cameraIdSet = new mutable.HashSet[String]()ele._2.foreach(row => cameraIdSet.add(row.getString(2)))//拼接字符串val info: String = Constants.FIELD_MONITOR_ID + "=" + monitorId + "|" + Constants.FIELD_AREA_ID + "=浦东新区|" + Constants.FIELD_CAMERA_IDS + "=" + cameraIdSet.mkString("-") + "|" + Constants.FIELD_CAMERA_COUNT + "=" + cameraIdSet.size + "|" + Constants.FIELD_CAR_COUNT + "=" + ele._2.size//返回结果(monitorId, info)})//-----------------------开始操作摄像头数据val monitorInfo: RDD[(String, String)] = sparkSession.sql("select * from monitor_camera_info").rdd.map(row => (row.getString(0), row.getString(1))).groupByKey().map(ele => {val monitorId: String = ele._1//拼接字符串val info: String = Constants.FIELD_CAMERA_IDS + "=" + ele._2.toList.mkString("-") + "|" + Constants.FIELD_CAMERA_COUNT + "=" + ele._2.size//返回结果(monitorId, info)//-----------------------将数据Join到一起monitorInfo.leftOuterJoin(flowInfo).foreach(println)})

在这里插入图片描述

3.3 车流量最多的TopN卡口

//开始计算val fRdd: RDD[Row] = sparkSession.sql("select * from monitor_flow_action where date = '2021-08-23' ").rddfRdd.map(row => (row.getString(7) + "_" + row.getString(6) + "&" + (Math.random() * 30 + 10).toInt, 1)).reduceByKey(_ + _).map(ele => {val area_road_random = ele._1val count = ele._2(area_road_random.split("_")(0), area_road_random.split("_")(1).split("&")(0) + "_" + count)}).groupByKey().map(ele => {val map = new mutable.HashMap[String, Int]()ele._2.foreach(e => {val key = e.split("_")(0)val value = e.split("_")(1).toIntmap.put(key, map.get(key).getOrElse(0) + value)})"区划【" + ele._1 + "】车辆最多的三条道路分别为:" + map.toList.sortBy(_._2).takeRight(3).reverse.mkString("-")}).foreach(println)

在这里插入图片描述

3.4 区域各路速度

随机抽取N个车辆信息，对这些数据可以进行多维度分析（因为随机抽取出来的N个车辆信息可以很权威的代表整个区域的车辆）

val sRdd: RDD[Row] = sparkSession.sql("select * from monitor_flow_action where date = '2021-08-23' ").rddsRdd.map(e=>{((e.getString(7),e.getString(6)),e.getString(5).toInt)}).groupByKey().map(e=>{val list: List[Int] = e._2.toListval i: Int = list.sum/list.size(e._1._1,(e._1._2,i))}).groupByKey().map(e=>{val tuples = e._2.toList.sortBy(_._2).reverse.take(3)var strBui: StringBuilder = new StringBuilderfor (i <- tuples ){val str: String = i._1 + "-均速度为：" + i._2strBui.append(">>>"+str)}(e._1,strBui)}).foreach(println)

在这里插入图片描述

3.5 区域中高速数量

在这里插入图片描述

object Hello04MonitorTopNSpeed {def main(args: Array[String]): Unit = {val sparkSession = ContextUtils.getSparkSession("Hello04MonitorTopNSpeed")MockDataUtil.mock2view(sparkSession)//---------------------开始操作车流量信息，假设任务编号为1 日期参数为今天val flowRdd: RDD[Row] = sparkSession.sql("select * from " + MockDataUtil.MONITOR_FLOW_ACTION + " where date = '2021-08-20' ").rddval monitor2speedRDD: RDD[(String, Iterable[String])] = flowRdd.map(row => (row.getString(1), row.getString(5))).groupByKey()val speedCount2monitorRDD: RDD[(SpeedCount, String)] = monitor2speedRDD.map(ele => {//获取卡口号val monitorId: String = ele._1//声明一个Map[0,60,100,120]var high = 0;var normal = 0;var low = 0;//获取所有的速度的车辆技术ele._2.foreach(speed => {//判断速度if (speed.toInt > 100) {high += 1} else if (speed.toInt > 60) {normal += 1} else {low += 1}})//创建速度对象(SpeedCount(high, normal, low), monitorId)})speedCount2monitorRDD.sortByKey(false).map(x => (x._2, x._1)).foreach(println)}
}case class SpeedCount(high: Int, normal: Int, low: Int) extends Ordered[SpeedCount] with KryoRegistrator {override def compare(that: SpeedCount): Int = {var result = this.high - that.highif (result == 0) {result = this.normal - that.normalif (result == 0) {result = this.low - that.low}}return result}override def registerClasses(kryo: Kryo): Unit = {kryo.register(SpeedCount.getClass)}
}

在这里插入图片描述

3.6 指定卡口对应卡口车辆轨迹

在这里插入图片描述

def main(args: Array[String]): Unit = {val sparkSession = ContextUtils.getSparkSession("Hello04MonitorTopNSpeed")MockDataUtil.mock2view(sparkSession)//获取数据val area01Rdd: RDD[Row] = sparkSession.sql("select * from " + MockDataUtil.MONITOR_FLOW_ACTION + " where date = '2021-08-23' and area_id = '01' ").rddval area02Rdd: RDD[Row] = sparkSession.sql("select * from " + MockDataUtil.MONITOR_FLOW_ACTION + " where date = '2021-08-23' and area_id = '02' ").rddval area01CarRdd = area01Rdd.map(row => (row.getString(3), row.getString(7))).groupByKey()val area02CarRdd = area02Rdd.map(row => (row.getString(3), row.getString(7))).groupByKey()area01CarRdd.join(area02CarRdd).foreach(println)}

在这里插入图片描述

3.7 行车轨迹

在这里插入图片描述

 def main(args: Array[String]): Unit = {    val sparkSession = ContextUtils.getSparkSession("AreaCar")    MockDataUtil.mock2view(sparkSession)//查询 车子行驶轨迹 跟车分析    val c1Rdd: RDD[Row] = sparkSession.sql("select * from " + MockDataUtil.MONITOR_FLOW_ACTION + " where date = '2021-08-23' ").rdd    val carRdd: RDD[(String, StringBuilder)] = c1Rdd.map(e => {      (e.getString(3), (e.getString(4), e.getString(6), e.getString(2)))    }).groupByKey()      .map(e => {        val tuples: List[(String, String, String)] = e._2.toList.sortBy(_._1)        val list = new StringBuilder        for (i <- tuples) {          //println(i)          val str: String = i._2 + ":" + i._3          list.append(str + "-")        }        (e._1, list)      })    //carRdd.foreach(println) 
}

在这里插入图片描述

3.9 车辆套牌

在这里插入图片描述

  def main(args: Array[String]): Unit = {val sparkSession = ContextUtils.getSparkSession("AreaCar")MockDataUtil.mock2view(sparkSession)
//假设任何的卡口距离都是 10分钟车程 ，如果同一分钟出现在不同的卡口就怀疑是套牌val deckRdd: RDD[Row] = sparkSession.sql("select * from " + MockDataUtil.MONITOR_FLOW_ACTION + " where date = '2021-08-23' ").rdddeckRdd.map(e => {val dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")(e.getString(3), (dateFormat.parse(e.getString(4)),e.getString(1)))}).groupByKey(1).map(e => {val list: List[(util.Date, String)] = e._2.toList.sortBy(x=>x._1)var bool = falsevar d: util.Date = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2021-08-23 00:00:00")var mid="?"for (i <- list) {if (d.getTime - i._1.getTime < 600000 && i._2!=mid )bool = trued = i._1mid=i._2}(e._1, bool)}).filter(f => f._2).foreach(println)}

在这里插入图片描述

3.10 车辆抽样-蓄水池抽样法

在这里插入图片描述

    def main(args: Array[String]): Unit = {val sparkSession = ContextUtils.getSparkSession("Hello04MonitorTopNSpeed")MockDataUtil.mock2view(sparkSession)//获取数据val flowRdd: RDD[Row] = sparkSession.sql("select * from " + MockDataUtil.MONITOR_FLOW_ACTION + " where date = '2021-08-21' ").rdd//yyyy-MM-dd_HH , rowval hourRDD: RDD[(String, Row)] = flowRdd.map(row => (DateUtils.getDateHour(row.getString(4)), row))//车流量的总数,并进行广播val flowAllCount: Long = hourRDD.count()val broadcastFlowAllCount: Broadcast[Long] = sparkSession.sparkContext.broadcast(flowAllCount)//计算每个小时的比例 并进行广播val hourRatio: collection.Map[String, Double] = hourRDD.countByKey().map(e => {(e._1, e._2 * 1.0 / broadcastFlowAllCount.value)})val broadcastHourRatio: Broadcast[collection.Map[String, Double]] = sparkSession.sparkContext.broadcast(hourRatio)//开始进行抽样val sampleRDD: RDD[Row] = hourRDD.groupByKey().flatMap(ele => {val hour: String = ele._1val list: List[Row] = ele._2.iterator.toList//计算本时段要抽样的数据量val sampleRatio: Double = broadcastHourRatio.value.get(hour).getOrElse(0)val sampleNum: Long = Math.round(sampleRatio * 100)//开始进行取样(蓄水池抽样)val sampleList: ListBuffer[Row] = new ListBuffer[Row]()sampleList.appendAll(list.take(sampleNum.toInt))for (i <- sampleNum until list.size) {//随机生成一个数字val num = (Math.random() * list.size).toIntif (num < sampleNum) {sampleList.update(num, list(i.toInt))}}sampleList})sampleRDD.foreach(println)}

在这里插入图片描述

3.11 道路转换率

在这里插入图片描述

  def main(args: Array[String]): Unit = {//创建会话val sparkSession = ContextUtils.getSparkSession("Hello07MonitorConvertRatio")MockDataUtil.mock2view(sparkSession)//开始计算val flowRdd: RDD[Row] = sparkSession.sql("select * from " + MockDataUtil.MONITOR_FLOW_ACTION + " where date = '2021-08-23' ").rdd//计算每个卡口的总通车量val monitorCountMap: collection.Map[String, Long] = flowRdd.map(row => (row.getString(1), row)).countByKey()//计算卡口到卡口的通行率val sortRDD: RDD[(String, List[Row])] = flowRdd.map(row => (row.getString(3), row)).groupByKey().map(ele => (ele._1, ele._2.iterator.toList.sortBy(_.getString(4))))val m2mMap: collection.Map[String, Long] = sortRDD.flatMap(ele => {//存放映射关系val map: mutable.HashMap[String, Int] = mutable.HashMap[String, Int]()val list: List[Row] = ele._2.toListfor (i <- 0 until list.size; j <- i + 1 until list.size) {//拼接Keyval key = list(i).getString(1) + "->" + list(j).getString(1)map.put(key, map.get(key).getOrElse(0) + 1);}//返回结果map.toList}).countByKey()//开始进行计算m2mMap.foreach(ele => {println("卡口[" + ele._1 + "]的转换率为:" + ele._2.toDouble / monitorCountMap.get(ele._1.split("->")(0)).get)})
}

在这里插入图片描述

3.12 区域通过的TopN卡口

  def main(args: Array[String]): Unit = {//创建会话val sparkSession = ContextUtils.getSparkSession("Hello07MonitorConvertRatio")MockDataUtil.mock2view(sparkSession)//开始计算val flowRdd: RDD[Row] = sparkSession.sql("select * from " + MockDataUtil.MONITOR_FLOW_ACTION + " where date = '2021-08-23' ").rdd//开始计算flowRdd.map(row => (row.getString(7) + "_" + row.getString(6) + "&" + (Math.random() * 30 + 10).toInt, 1)).reduceByKey(_ + _).map(ele => {val area_road_random = ele._1val count = ele._2(area_road_random.split("_")(0), area_road_random.split("_")(1).split("&")(0) + "_" + count)}).groupByKey().map(ele => {val map = new mutable.HashMap[String, Int]()ele._2.foreach(e => {val key = e.split("_")(0)val value = e.split("_")(1).toIntmap.put(key, map.get(key).getOrElse(0) + value)})"区划【" + ele._1 + "】车辆最多的三条道路分别为:" + map.toList.sortBy(_._2).takeRight(3).reverse.mkString("-")}).foreach(println)
}
areaId2DetailInfos"SELECT ""monitor_id,""car,""road_id,""area_id ""FROM traffic.monitor_flow_action ""WHERE date >= '"startDate"'""AND date <= '"endDate"'"areaId2AreaInfoRDDareaid          areanametmp_car_flow_basic = monitor_flow_action  areaId2AreaInfoRDDmonitor_id car road_id  area_id area_name 统计各个区域各个路段车流量的临时表area_name  road_id    car_count      monitor_infos海淀区    01             100      0001=20|0002=30|0003=50注册成临时表tmp_area_road_flow_count"SELECT ""area_name,""road_id,""count(*) car_count,"//group_concat_distinct 统计每一条道路中每一个卡扣下的车流量"group_concat_distinct(monitor_id) monitor_infos "//0001=20|0002=30"FROM tmp_car_flow_basic ""GROUP BY area_name,road_id"0001=20|0002=30insert into areaTop3Road"SELECT ""area_name,""road_id,""car_count,""monitor_infos, ""CASE ""WHEN car_count > 170 THEN 'A LEVEL' ""WHEN car_count > 160 AND car_count <= 170 THEN 'B LEVEL' ""WHEN car_count > 150 AND car_count <= 160 THEN 'C LEVEL' ""ELSE 'D LEVEL' ""END flow_level "
"FROM (""SELECT ""area_name,""road_id,""car_count,""monitor_infos,""row_number() OVER (PARTITION BY area_name ORDER BY car_count DESC) rn ""FROM tmp_area_road_flow_count "") tmp "
"WHERE rn <=3"

这篇关于Spark项目实战-卡口流量统计的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/283064。

相关文章

基于 Cursor 开发 Spring Boot 项目详细攻略

基于 Cursor 开发 Spring Boot 项目详细攻略

《基于Cursor开发SpringBoot项目详细攻略》Cursor是集成GPT4、Claude3.5等LLM的VSCode类AI编程工具,支持SpringBoot项目开发全流程,涵盖环境配... 目录cursor是什么？基于 Cursor 开发 Spring Boot 项目完整指南1. 环境准备2. 创建
阅读更多...
MyBatis分页查询实战案例完整流程

MyBatis分页查询实战案例完整流程

《MyBatis分页查询实战案例完整流程》MyBatis是一个强大的Java持久层框架,支持自定义SQL和高级映射,本案例以员工工资信息管理为例,详细讲解如何在IDEA中使用MyBatis结合Page... 目录1. MyBATis框架简介2. 分页查询原理与应用场景2.1 分页查询的基本原理2.1.1 分
阅读更多...
使用Python批量将.ncm格式的音频文件转换为.mp3格式的实战详解

使用Python批量将.ncm格式的音频文件转换为.mp3格式的实战详解

《使用Python批量将.ncm格式的音频文件转换为.mp3格式的实战详解》本文详细介绍了如何使用Python通过ncmdump工具批量将.ncm音频转换为.mp3的步骤,包括安装、配置ffmpeg环... 目录1. 前言2. 安装 ncmdump3. 实现 .ncm 转 .mp34. 执行过程5. 执行结
阅读更多...
C++统计函数执行时间的最佳实践

C++统计函数执行时间的最佳实践

《C++统计函数执行时间的最佳实践》在软件开发过程中,性能分析是优化程序的重要环节,了解函数的执行时间分布对于识别性能瓶颈至关重要,本文将分享一个C++函数执行时间统计工具,希望对大家有所帮助... 目录前言工具特性核心设计1. 数据结构设计2. 单例模式管理器3. RAII自动计时使用方法基本用法高级用法
阅读更多...
SpringBoot 多环境开发实战(从配置、管理与控制)

SpringBoot 多环境开发实战(从配置、管理与控制)

《SpringBoot多环境开发实战(从配置、管理与控制)》本文详解SpringBoot多环境配置,涵盖单文件YAML、多文件模式、MavenProfile分组及激活策略,通过优先级控制灵活切换环境... 目录一、多环境开发基础（单文件 YAML 版）（一）配置原理与优势（二）实操示例二、多环境开发多文件版
阅读更多...
Three.js构建一个 3D 商品展示空间完整实战项目

Three.js构建一个 3D 商品展示空间完整实战项目

《Three.js构建一个3D商品展示空间完整实战项目》Three.js是一个强大的JavaScript库,专用于在Web浏览器中创建3D图形,：本文主要介绍Three.js构建一个3D商品展... 目录引言项目核心技术1. 项目架构与资源组织2. 多模型切换、交互热点绑定3. 移动端适配与帧率优化4. 可
阅读更多...
sky-take-out项目中Redis的使用示例详解

sky-take-out项目中Redis的使用示例详解

《sky-take-out项目中Redis的使用示例详解》SpringCache是Spring的缓存抽象层,通过注解简化缓存管理,支持Redis等提供者,适用于方法结果缓存、更新和删除操作,但无法实现... 目录Spring Cache主要特性核心注解1.@Cacheable2.@CachePut3.@Ca
阅读更多...
从原理到实战解析Java Stream 的并行流性能优化

从原理到实战解析Java Stream 的并行流性能优化

《从原理到实战解析JavaStream的并行流性能优化》本文给大家介绍JavaStream的并行流性能优化：从原理到实战的全攻略,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的... 目录一、并行流的核心原理与适用场景二、性能优化的核心策略1. 合理设置并行度：打破默认阈值2. 避免装箱
阅读更多...
Maven中生命周期深度解析与实战指南

Maven中生命周期深度解析与实战指南

《Maven中生命周期深度解析与实战指南》这篇文章主要为大家详细介绍了Maven生命周期实战指南,包含核心概念、阶段详解、SpringBoot特化场景及企业级实践建议,希望对大家有一定的帮助... 目录一、Maven 生命周期哲学二、default生命周期核心阶段详解（高频使用）三、clean生命周期核心阶
阅读更多...
Python实战之SEO优化自动化工具开发指南

Python实战之SEO优化自动化工具开发指南

《Python实战之SEO优化自动化工具开发指南》在数字化营销时代,搜索引擎优化（SEO）已成为网站获取流量的重要手段,本文将带您使用Python开发一套完整的SEO自动化工具,需要的可以了解下... 目录前言项目概述技术栈选择核心模块实现1. 关键词研究模块2. 网站技术seo检测模块3. 内容优化分析模
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2