从基础的道路通行状态统计、分析、聚类等维度开展对某个城市道路拥堵情况的分析和研究。
13.3.2 数据预处理根据给定的某地图路况数据，首先进行数据预处理工作，清洗原始数据并去除缺失数据、错误数据，根据道路ID进行数据汇集，计算拥堵指数。1）清除缺失数据：清除字段为空记录；
2）清除错误数据：清除字段错误记录；
3）根据道路ID进行道路拥堵指数聚合；
4）根据时间进行道路拥堵指数排序。
13.3.3 特征构建仍然以半小时为最小时间粒度（每日24小时划分为48维时间片），并对道路拥堵指数按时间片进行聚合计算，同时按照48维时间片进行拥堵指数排列。具体处理过程以及代码如下：
package com.koala.ch13
import org.apache.spark.{SparkConf, SparkContext}
import java.text.SimpleDateFormat
import java.util.Calendar
import breeze.linalg.Counter
import org.apache.log4j.{Level, Logger}
object CrowdModel {
def main(args: Array[String]){
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
if (args.length < 2) {
System.err.println("Usage:CrowdModel <InPath> <OutPut> <Model>")
System.exit(1)
}
// 2rd_data/ch13/user_location_sample.txt output/ch13/CrowdModel local[2]
val Array(input,output,mode) = args
//初始化SparkContext
val conf = new SparkConf()
.setMaster(mode)//.setMaster("local")
.setAppName(this.getClass.getSimpleName)
val sc = new SparkContext(conf)
// 位置筛选
// 清洗数据，通过split(",")切分数据，得到 User_id Time_stamp Cell_id三个维度的数据列表。
// (Time_stamp,Cell_id,User_id)-> (User_id,Time_stamp,Cell_id)
// 20160501055052,209059,898
val data = sc.textFile(input).map(_.split(",")).map {
x => (x(2), x(0), x(1))
}
//data.coalesce(1).saveAsTextFile(output)
// 根据Time_stamp分析当前日期为工作日或节假日，并添加time标签标识HH:mm，work_flag标签标识工作日（work_falg=1）或节假日（work_flag=0）
// 输出：(User_id,work_flag,date_time,Cell_id)
val preData = data.map {
case (preUser_id, preTime_stamp, preCell_id) => {
//将日期转变成星期，获取工作日（星期一至星期五）和非工作日（星期六、星期日）
// 20160501 055052
val sdf = new SimpleDateFormat("yyyyMMddHHmmss") //24小时工作制
val date = sdf.parse(preTime_stamp)
val cal = Calendar.getInstance
cal.setTime(date)
var w = cal.get(Calendar.DAY_OF_WEEK) - 1
// 工作日默认为1 非工作日默认为0
var work_flag = 1
if (w <= 0 || w >= 6) {
work_flag = 0
}
// 按照30分钟间隔处理时间
val time_ = preTime_stamp.substring(8, 12)
// 截取指定位置的元素，前包括后不包括
var minute_ = "00"
if (time_.substring(2).toInt >= 30) {
minute_ = "30"
}
val date_time = time_.toString.substring(0, 2) + minute_
((preUser_id, work_flag, date_time, preCell_id), 1)
}
}
//preData.coalesce(1).saveAsTextFile(output)
//使用reduceByKey(_+_)对(User_id,work_flag,date_time,Cell_id)访问次数进行聚合，根据聚合结果，选择用户某段时间在30分钟内划分访问次数最多的基站为标准访问地点。
val aggData = preData.reduceByKey(_ + _)
.map { x => ((x._1._1, x._1._2, x._1._3), (x._1._4, x._2)) }
.reduceByKey((a, b) => if (a._2 > b._2) a else b)//选取访问次数最多的cell
//aggData.coalesce(1).saveAsTextFile(output)
//获取用户工作日24小时访问地点cell_id、节假日24小时访问地点cell_id，以30分钟为最小时间粒度划分时间片，得到user_id工作日48维时间片访问cell_id和节假日48维时间片访问cell_id，共计96维时间片。
//(User_id,work_flag,date_time),(Cell_id,nums)->(User_id,work_flag),(date_time,Cell_id)
val slotData = aggData.map { x => ((x._1._1, x._1._2), (x._1._3 + ":" + x._2._1)) }.reduceByKey(_ + ";" + _)
//slotData.coalesce(1).saveAsTextFile(output)
// 位置编码
// 根据聚合结果，提取所有用户访问的基站进行重新编码，获得用户访问位置列表cell_id，并进行排序去重
// (User_id,work_flag,date_time),(Cell_id,nums)
val minCell = aggData.map(x => x._2._1).sortBy(x=>x.toLong,true).collect().distinct
println(minCell.toList)
//使用zip方法从1开始对用户访问地址进行编码，并将编码进行保存。
val index_list = minCell.zip(Stream from 1).toMap
println(index_list)
//得到的index_list即是用户访问位置编码特征向量。
}
}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
object CleanCongestionData {
def main(args: Array[String]) {
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
if (args.length < 2) {
System.err.println("Usage:CleanCongestionData <InPath> <OutPut> <Model>")
System.exit(1)
}
// 2rd_data/ch13/road_congestion_sample.txt output/ch13/CongestionModel local[2]
val Array(input,output,mode) = args
// 初始化SparkContext
val conf = new SparkConf()
.setMaster(mode)//.setMaster("local")
.setAppName(this.getClass.getSimpleName)
val sc = new SparkContext(conf)
// 计算link的拥堵情况，指定道路、工作日状态、时间片时，link拥堵指数的平均值（四舍五入）取整，
// key (linkid, work_flag, hour) value (congestion)
// 85349482;1;20.5;1
val data = sc.textFile(input).map(_.split(";"))
.map {x => ((x(0),x(1),x(2)),x(3))}
.groupByKey().mapValues(x=>{
val a = x.toList.reduceLeft((sum,i)=>sum +i)//拥堵指数求和
val b = x.toList.length
Math.round(a.toInt/b)//平均拥堵指数
})
//data.coalesce(1).saveAsTextFile(output)
// 根据key聚合数据后，使用hour 进行排序 并删除hour数据
// key (linkid,work_flag, hour) value (congestion)->(linkid) value(work_flag,congestion)
val collectData = data.sortBy(x=>x._1._3).map(x => ((x._1._1),(x._1._2+":"+x._2))).reduceByKey(_ + ";" + _)
collectData.coalesce(1).saveAsTextFile(output)
}
}