12.5 地图实时路况数据分析

以你之姓@ 2022-05-15 09:34 288阅读 0赞

运行之前：启动zookeeper，kafka，redis

注意jar兼容：代码：Traffic.zip

![70][]

地图实时路况：

可以使用Java代码来当kafka生产者

def main(args: Array\[String\]): Unit = \{

val topic = "car\_events"

val brokers = "node2:9092,node3:9092,node4:9092"

val props = new Properties()

props.put("metadata.broker.list", brokers)

props.put("serializer.class", "kafka.serializer.StringEncoder")

val kafkaConfig = new ProducerConfig(props)

val producer = new Producer\[String, String\](kafkaConfig)

val sparkConf = new SparkConf().setAppName("Beijing traffic").setMaster("local\[4\]")

val sc = new SparkContext(sparkConf)

// val filePath = "D:/traffic/trafficlf\_all\_column\_all.txt"

val filePath = "data/2014082013\_all\_column\_test.txt"

val records = sc.textFile(filePath)

.filter(!\_.startsWith(";"))

.map(\_.split(",")).collect()

for(record <- records)\{

// prepare event data

val event = new JSONObject()

event

.put("camera\_id", record(0))

.put("car\_id", record(2))

.put("event\_time", record(4))

.put("speed", record(6))

.put("road\_id", record(13))

// produce event message

producer.send(new KeyedMessage\[String, String\](topic, event.toString))

// producer.send(new KeyedMessage\[String, String\](topic, "yuntian",event.toString))

println("Message sent: " + event)

Thread.sleep(200)

--------------------

代码当消费者，并存储到redis数据库：

def main(args: Array\[String\]): Unit = \{

var masterUrl = "local\[1\]"

if (args.length > 0) \{

masterUrl = args(0)

val conf = new SparkConf().setMaster(masterUrl).setAppName("UserClickCountStat")

val ssc = new StreamingContext(conf, Seconds(5))

val topics = Set("car\_events")

val brokers = "node2:9092,node3:9092,node4:9092"

val kafkaParams = Map\[String, String\](

"metadata.broker.list" -> brokers, "serializer.class" -> "kafka.serializer.StringEncoder")

val dbIndex = 1

val kafkaStream = KafkaUtils.createDirectStream\[String, String, StringDecoder, StringDecoder\](ssc, kafkaParams, topics)

val events = kafkaStream.flatMap(line => \{

val data = JSONObject.fromObject(line.\_2)

println(data)

Some(data)

\})

// (Some(data)) --> (camera\_id, speed)

val carSpeed = events.map(x => (x.getString("camera\_id"),x.getInt("speed")))

// val carSpeed = events.map(x => x.getString("road\_id") -> (x.getInt("speed"),1))

// .reduceByKey((a, b) => \{(a.\_1 + b.\_1, a.\_2 + b.\_2)\})

.mapValues((x:Int)=>(x,1.toInt))

// .reduceByKeyAndWindow((a, b) => \{(a.\_1 + b.\_1, a.\_2 + b.\_2)\},Seconds(10))

// (camera\_id,(speed,1))

.reduceByKeyAndWindow((a:Tuple2\[Int,Int\], b:Tuple2\[Int,Int\]) => \{(a.\_1 + b.\_1, a.\_2 + b.\_2)\},Seconds(20),Seconds(10))

// carSpeed.map\{ case (key, value) => (key, value.\_1 / value.\_2.toFloat) \}

carSpeed.foreachRDD(rdd => \{

rdd.foreachPartition(partitionOfRecords => \{

val jedis = RedisClient.pool.getResource

// (camera\_id,(speed,1))

partitionOfRecords.foreach(pair => \{

val camera\_id = pair.\_1

val total = pair.\_2.\_1

val count = pair.\_2.\_2

val now = Calendar.getInstance().getTime()

// create the date/time formatters

val minuteFormat = new SimpleDateFormat("HHmm")

val dayFormat = new SimpleDateFormat("yyyyMMdd")

val time = minuteFormat.format(now)

val day = dayFormat.format(now)

if(count!=0)\{

// val averageSpeed = total / count

jedis.select(dbIndex)//连接redis数据库,dbIndex设置连接几号数据库

jedis.hset(day + "\_" + camera\_id, time , total + "\_" + count)

// fetch data from redis

// val temp = jedis.hget(day + "\_" + camera\_id, time)

// println(temp)

\})

//RedisClient里面封装了redis数据库信息

RedisClient.pool.returnResource(jedis)

\})

println("xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx")

ssc.start()

ssc.awaitTermination()

--------------------

RedisClient.scala：

object RedisClient extends Serializable \{

val redisHost = "node2"

val redisPort = 6379

val redisTimeout = 30000

lazy val pool = new JedisPool(new GenericObjectPoolConfig(), redisHost, redisPort, redisTimeout)

lazy val hook = new Thread \{

override def run = \{

println("Execute hook thread: " + this)

pool.destroy()

sys.addShutdownHook(hook.run)

[70]: /images/20220515/5d2f67537865404b9eba07710945fa9f.png

12.5 地图实时路况数据分析

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 125.树形数据分级汇总

相关 Tableau数据分析-Chapter07多边形地图和背景地图

相关 Tableau数据分析-Chapter06填充地图、多维地图、混合地图

相关实时日活数据分析

相关【百度地图】获取实时路况地图瓦片

相关高德地图之实时导航

相关 12.6 地图实时路况预测

相关 12.5 地图实时路况数据分析

相关地图点儿莱唔点儿康的新功能 – 路况和短信

相关百度地图-大数据量点实时更新