一、车辆超速信息监控

  • 从kafka读取卡口车辆的速度信息，关联卡口的限速信息表，统计车辆超速20%车辆，保存到mysql结果表中。
  • 注意：使用广播数据流，使用场景，两个流要关联，一个流大，一个流小，小的流会定期变化，可以将小的流全量广播到每个taskManager中。
  • 主要涉及kafkasource、mysqlsink、广播算子的样例代码 ```scala import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet} import java.util.Properties

import com.bjsxt.traffic.utils.{JDBCSink, MonitorCarInfo, MonitorLimitSpeedInfo, OverSpeedCarInfo} import org.apache.flink.api.common.serialization.SimpleStringSchema import org.apache.flink.api.common.state.{BroadcastState, MapStateDescriptor, ReadOnlyBroadcastState} import org.apache.flink.configuration.Configuration import org.apache.flink.streaming.api.datastream.BroadcastStream import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction} import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, FlinkKafkaConsumerBase} import org.apache.flink.util.Collector import org.apache.kafka.common.serialization.StringSerializer

object OverSpeedCarAnaly { def main(args: Array[String]): Unit = { val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment import org.apache.flink.streaming.api.scala._

  1. //设置Kafka 配置
  2. val props = new Properties()
  3. props.setProperty("bootstrap.servers","mynode1:9092,mynode2:9092,mynode3:9092")
  4. props.setProperty("key.deserializer",classOf[StringSerializer].getName)
  5. props.setProperty("value.deserializer",classOf[StringSerializer].getName)
  6. props.setProperty("group.id","group1125")
  7. //创建Kafka Source，monitortopic1125表示读取的topic
  8. //setStartFromEarliest表示从最开始读取数据
  9. val kafkaSource: FlinkKafkaConsumerBase[String] = new FlinkKafkaConsumer[String]("monitortopic1125",new SimpleStringSchema(),props).setStartFromEarliest()
  10. //读取车辆监控信息
  11. val ds1: DataStream[String] = env.addSource(kafkaSource)
  12. //    val ds1: DataStream[String] = env.socketTextStream("mynode5",9999)
  14. //对车辆监控信息进行转换数据
  15. //MonitorCarInfo格式如下：
  16. //areaId:String,roadId:String,monitorId:String,cameraId:String,actionTime:Long,car:String,speed:Double
  17. val carMonitorDS: DataStream[MonitorCarInfo] = ds1.map(line => {
  18.   val arr: Array[String] = line.split("\t")
  19.   MonitorCarInfo(arr(0), arr(1), arr(2), arr(3), arr(4).toLong, arr(5), arr(6).toDouble)
  20. })
  22. //读取区域-道路-卡扣 车辆限速信息，读取mysql中的数据，每隔半小时读取一次
  23. //使用 RichSourceFunction 生成卡口限速信息
  24. val limitSpeedDs: DataStream[MonitorLimitSpeedInfo] = env.addSource(new RichSourceFunction[MonitorLimitSpeedInfo] {
  26.   var conn: Connection = _
  27.   var pst: PreparedStatement = _
  28.   var stop = false
  30.   //当初始化 RichSourceFunction时调用一次，可以初始化数据库连接对象
  31.   override def open(parameters: Configuration): Unit = {
  32.     conn = DriverManager.getConnection("jdbc:mysql://192.168.179.14:3306/traffic_monitor", "root", "123456")
  33.     pst = conn.prepareStatement("select area_id,road_id,monitor_id,speed_limit from t_monitor_speed_limit_info")
  34.   }
  36.   //产生数据的方法，一般使用一个循环来实现读取mysql数据
  37.   override def run(ctx: SourceFunction.SourceContext[MonitorLimitSpeedInfo]): Unit = {
  38.     while (!stop) {
  39.       val set: ResultSet = pst.executeQuery()
  40.       while (set.next()) {
  41.         val areaId: String = set.getString("area_id")
  42.         val roadId: String = set.getString(2)
  43.         val monitorId: String = set.getString(3)
  44.         val speedLimit: Double = set.getDouble(4)
  45.         ctx.collect(new MonitorLimitSpeedInfo(areaId, roadId, monitorId, speedLimit))
  46.       }
  48.       //每隔半小时查询一次数据库全量的限速信息
  49.       Thread.sleep(30 * 1000 * 60)
  50.     }
  51.   }
  53.   //当Flink程序取消时，自动调用cancel方法
  54.   override def cancel(): Unit = {
  55.     stop = true
  56.     pst.close()
  57.     conn.close()
  58.   }
  59. })
  61. //将从mysql中读取的限速信息，使用广播流广播出去
  62. val mapStateDesc = new MapStateDescriptor[String,Double]("map-state",classOf[String],classOf[Double])
  63. val bcLimitSpeed: BroadcastStream[MonitorLimitSpeedInfo] = limitSpeedDs.broadcast(mapStateDesc)
  65. //连接车辆监控的数据流与当前的广播流 ,进行车辆超速判断
  66. val overSpeedInfoDs: DataStream[OverSpeedCarInfo] = carMonitorDS.connect(bcLimitSpeed).process(new BroadcastProcessFunction[MonitorCarInfo, MonitorLimitSpeedInfo, OverSpeedCarInfo] {
  67.   //针对车辆监控数据进行处理，value : 当前本条数据 cxt：Flink上下文，out：回收数据对象
  68.   override def processElement(value: MonitorCarInfo, ctx: BroadcastProcessFunction[MonitorCarInfo, MonitorLimitSpeedInfo, OverSpeedCarInfo]#ReadOnlyContext, out: Collector[OverSpeedCarInfo]): Unit = {
  69.     val areaId: String = value.areaId
  70.     val monitorId: String = value.monitorId
  71.     val cameraId: String = value.cameraId
  72.     //本条数据区域_道路_卡扣 信息
  73.     val key = areaId + "_" + monitorId + "_" + cameraId
  74.     //获取广播状态，判断当前数据key 是否有对应的限速状态
  75.     val readOnlyBS: ReadOnlyBroadcastState[String, Double] = ctx.getBroadcastState(mapStateDesc)
  76.     if (readOnlyBS.contains(key)) {
  77.       //如果包含当前key对应的限速
  78.       if (value.speed > readOnlyBS.get(key) * 1.2) {
  79.         //超速车辆，回收回去
  80.         out.collect(new OverSpeedCarInfo(value.car, value.monitorId, value.roadId, value.speed, readOnlyBS.get(key), value.actionTime))
  81.       }
  82.     } else {
  83.       //如果不包含限速信息，默认限速为60
  84.       if (value.speed > 60 * 1.2) {
  85.         //超速车辆信息，回收回去
  86.         out.collect(new OverSpeedCarInfo(value.car, value.monitorId, value.roadId, value.speed, 60.0, value.actionTime))
  87.       }
  88.     }
  89.   }
  91.   //针对广播流数据进行处理 ，value :当前广播的此条数据，ctx :Flink上下文，out：回收数据对象
  92.   override def processBroadcastElement(value: MonitorLimitSpeedInfo, ctx: BroadcastProcessFunction[MonitorCarInfo, MonitorLimitSpeedInfo, OverSpeedCarInfo]#Context, out: Collector[OverSpeedCarInfo]): Unit = {
  93.     val broadCastState: BroadcastState[String, Double] = ctx.getBroadcastState(mapStateDesc)
  94.     val key = value.areaId + "_" + value.roadId + "_" + value.monitorId //key : 区域_道路_卡扣
  95.     broadCastState.put(key, value.limitSpeed)
  96.   }
  97. })
  99. //将结果写入mysql 表 t_speed_info
  100. overSpeedInfoDs.addSink(new JDBCSink[OverSpeedCarInfo]("OverSpeedCarInfo"))
  102. env.execute()

} }

import java.util import org.apache.flink.configuration.Configuration import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction} import org.apache.hadoop.hbase.HBaseConfiguration import org.apache.hadoop.hbase.client.{ConnectionFactory, Put, Table} import org.apache.hadoop.conf import org.apache.hadoop.hbase.{HBaseConfiguration, TableName, client} /**

  • HBase sink ,批量数据插入到HBase中 */ class HBaseSink extends RichSinkFunction[java.util.List[Put]] { var configuration: conf.Configuration = var conn: client.Connection = //初始化 RichSinkFunction 对象时 执行一次 override def open(parameters: Configuration): Unit = { configuration = HBaseConfiguration.create() configuration.set(“hbase.zookeeper.quorum”,”mynode3:2181,mynode4:2181,mynode5:2181”) conn = ConnectionFactory.createConnection(configuration) }

    //每条数据执行一次 override def invoke(value: util.List[Put], context: SinkFunction.Context[_]): Unit = { //连接HBase 表 val table: Table = conn.getTable(TableName.valueOf(“a1”)) table.put(value) } } ```

二、每个区域每个道路每个卡扣的平均速度

  • 每隔1分钟统计过去5分钟，每个区域每个道路每个卡扣的平均速度
  • 涉及EventTime时间语义 ```scala import java.text.SimpleDateFormat import java.util.{Date, Properties}

import com.bjsxt.traffic.utils.{JDBCSink, MonitorAvgSpeedInfo, MonitorCarInfo} import org.apache.flink.api.common.functions.AggregateFunction import org.apache.flink.api.common.serialization.SimpleStringSchema import org.apache.flink.streaming.api.TimeCharacteristic import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment import org.apache.flink.streaming.api.scala.function.WindowFunction import org.apache.flink.streaming.api.windowing.time.Time import org.apache.flink.streaming.api.windowing.windows.TimeWindow import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer import org.apache.flink.util.Collector import org.apache.kafka.common.serialization.StringDeserializer

/**

  • 每隔1分钟统计过去5分钟，每个区域每个道路每个卡扣的平均速度 */ object MonitorAvgSpeedAnaly { def main(args: Array[String]): Unit = { val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment import org.apache.flink.streaming.api.scala._

    //设置并行度和事件时间 env.setParallelism(1) env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //读取Kafka中 实时监控车辆数据 val props = new Properties() props.setProperty(“bootstrap.servers”,”mynode1:9092,mynode2:9092,mynode3:9092”) props.setProperty(“key.deserializer”,classOf[StringDeserializer].getName) props.setProperty(“value.deserializer”,classOf[StringDeserializer].getName) props.setProperty(“group.id”,”group112502”)

    //读取Kafka 中监控到实时的车辆信息 val monitorInfosDs: DataStream[String] = env.addSource(new FlinkKafkaConsumerString,props).setStartFromEarliest()) // val monitorInfosDs: DataStream[String] = env.socketTextStream(“mynode5”,9999) //数据类型转换 val transferDS: DataStream[MonitorCarInfo] = monitorInfosDs.map(line => { val arr: Array[String] = line.split(“\t”) MonitorCarInfo(arr(0), arr(1), arr(2), arr(3), arr(4).toLong, arr(5), arr(6).toDouble) }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractorMonitorCarInfo) { override def extractTimestamp(element: MonitorCarInfo): Long = element.actionTime//设置EventTime的字段 })

  1. //设置窗口，每隔1分钟计算每个卡扣的平均速度,使用aggregate 增量计算+全量计算
  2. val monitorAvgSpeedDS: DataStream[MonitorAvgSpeedInfo] = transferDS.keyBy(mi => {
  3.   mi.areaId + "_" + mi.roadId + "_" + mi.monitorId
  4. }).timeWindow(Time.minutes(5), Time.minutes(1))
  5.   .aggregate(new AggregateFunction[MonitorCarInfo, (Long, Double), (Long, Double)] {
  6.     override def createAccumulator(): (Long, Double) = (0L, 0L)
  8.     override def add(value: MonitorCarInfo, accumulator: (Long, Double)): (Long, Double) = {
  9.       (accumulator._1 + 1, accumulator._2 + value.speed)
  10.     }
  12.     override def getResult(accumulator: (Long, Double)): (Long, Double) = accumulator
  14.     override def merge(a: (Long, Double), b: (Long, Double)): (Long, Double) = (a._1 + b._1, a._2 + b._2)
  15.   },
  16.     new WindowFunction[(Long, Double), MonitorAvgSpeedInfo, String, TimeWindow] {
  17.       //key : 区域_道路_卡扣  window : 当前窗口，input ：输入的数据 ，out:回收数据对象
  18.       override def apply(key: String, window: TimeWindow, input: Iterable[(Long, Double)], out: Collector[MonitorAvgSpeedInfo]): Unit = {
  19.         val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
  20.         val startTime: String = sdf.format(new Date(window.getStart))
  21.         val endTime: String = sdf.format(new Date(window.getEnd))
  22.         //计算当前卡扣的平均速度
  23.         val last: (Long, Double) = input.last
  24.         val avgSpeed: Double = (last._2 / last._1).formatted("%.2f").toDouble
  25.         out.collect(new MonitorAvgSpeedInfo(startTime, endTime, key, avgSpeed, last._1))
  26.       }
  27.     })
  29. //保存到mysql 数据库中
  30. monitorAvgSpeedDS.addSink(new JDBCSink[MonitorAvgSpeedInfo]("MonitorAvgSpeedInfo"))
  31. env.execute()

} }

//定义卡扣平均速度信息 case class MonitorAvgSpeedInfo(windowStartTime:String,windowEndTime:String,monitorId:String,avgSpeed:Double,carCount:Long)

  1. <a name="X3shh"></a>
  2. ## 三、每隔1分钟统计最近5分钟 最通畅的topN卡扣信息
  3. ```scala
  4. import java.util.Properties
  5. import com.bjsxt.traffic.utils._
  6. import org.apache.flink.api.common.serialization.SimpleStringSchema
  7. import org.apache.flink.streaming.api.TimeCharacteristic
  8. import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
  9. import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
  10. import org.apache.flink.streaming.api.scala.function.ProcessAllWindowFunction
  11. import org.apache.flink.streaming.api.windowing.time.Time
  12. import org.apache.flink.streaming.api.windowing.windows.TimeWindow
  13. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
  14. import org.apache.flink.util.Collector
  15. import org.apache.kafka.common.serialization.StringDeserializer
  17. object TopNMinAvgSpeedMonitorAnaly {
  18.   def main(args: Array[String]): Unit = {
  19.     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  20.     import org.apache.flink.streaming.api.scala._
  22.     //设置并行度和事件时间
  23.     env.setParallelism(1)
  24.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  26.     //读取Kafka中 实时监控车辆数据
  27.     val props = new Properties()
  28.     props.setProperty("bootstrap.servers","mynode1:9092,mynode2:9092,mynode3:9092")
  29.     props.setProperty("key.deserializer",classOf[StringDeserializer].getName)
  30.     props.setProperty("value.deserializer",classOf[StringDeserializer].getName)
  31.     props.setProperty("group.id","group112502")
  33.     //读取Kafka 中监控到实时的车辆信息
  34.     val monitorInfosDs: DataStream[String] = env.addSource(new FlinkKafkaConsumer[String]("monitortopic1125",new SimpleStringSchema(),props).setStartFromEarliest())
  35. //    val monitorInfosDs: DataStream[String] = env.socketTextStream("mynode5",9999)
  36.     //数据类型转换
  37.     val transferDS: DataStream[MonitorCarInfo] = monitorInfosDs.map(line => {
  38.       val arr: Array[String] = line.split("\t")
  39.       MonitorCarInfo(arr(0), arr(1), arr(2), arr(3), arr(4).toLong, arr(5), arr(6).toDouble)
  40.     }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[MonitorCarInfo](Time.seconds(5)) {
  41.       override def extractTimestamp(element: MonitorCarInfo): Long = element.actionTime
  42.     })
  44.     //每隔1分钟统计过去5分钟，最通畅的top5卡扣信息
  45.     val top5MonitorInfoDS: DataStream[Top5MonitorInfo] = transferDS.timeWindowAll(Time.minutes(5), Time.minutes(1))
  46.       .process(new ProcessAllWindowFunction[MonitorCarInfo, Top5MonitorInfo, TimeWindow] {
  47.         //context :Flink 上下文，elements : 窗口期内所有元素，out: 回收数据对象
  48.         override def process(context: Context, elements: Iterable[MonitorCarInfo], out: Collector[Top5MonitorInfo]): Unit = {
  49.           val map = scala.collection.mutable.Map[String, MonitorSpeedClsCount]()
  50.           val iter: Iterator[MonitorCarInfo] = elements.iterator
  51.           while (iter.hasNext) {
  52.             val currentInfo: MonitorCarInfo = iter.next()
  53.             val areaId: String = currentInfo.areaId
  54.             val roadId: String = currentInfo.roadId
  55.             val monitorId: String = currentInfo.monitorId
  56.             val speed: Double = currentInfo.speed
  57.             val currentKey = areaId + "_" + roadId + "_" + monitorId
  58.             //判断当前map中是否含有当前本条数据对应的 区域_道路_卡扣的信息
  59.             if (map.contains(currentKey)) {
  60.               //判断当前此条车辆速度位于哪个速度端，给map中当前key 对应的value MonitorSpeedClsCount 对象对应的速度段加1
  61.               if (speed >= 120) {
  62.                 map.get(currentKey).get.hightSpeedCarCount += 1
  63.               } else if (speed >= 90) {
  64.                 map.get(currentKey).get.middleSpeedCount += 1
  65.               } else if (speed >= 60) {
  66.                 map.get(currentKey).get.normalSpeedCarCount += 1
  67.               } else {
  68.                 map.get(currentKey).get.lowSpeedCarCount += 1
  69.               }
  70.             } else {
  71.               //不包含 当前key
  72.               val mscc = MonitorSpeedClsCount(0L, 0L, 0L, 0L)
  73.               if (speed >= 120) {
  74.                 mscc.hightSpeedCarCount += 1
  75.               } else if (speed >= 90) {
  76.                 mscc.middleSpeedCount += 1
  77.               } else if (speed >= 60) {
  78.                 mscc.normalSpeedCarCount += 1
  79.               } else {
  80.                 mscc.lowSpeedCarCount += 1
  81.               }
  82.               map.put(currentKey, mscc)
  83.             }
  84.           }
  86.           val tuples: List[(String, MonitorSpeedClsCount)] = map.toList.sortWith((tp1, tp2) => {
  87.             tp1._2 > tp2._2
  88.           }).take(5)
  89.           for (elem <- tuples) {
  90.             val windowStartTime: String = DateUtils.timestampToDataStr(context.window.getStart)
  91.             val windowEndTime: String = DateUtils.timestampToDataStr(context.window.getEnd)
  92.             out.collect(new Top5MonitorInfo(windowStartTime, windowEndTime, elem._1, elem._2.hightSpeedCarCount, elem._2.middleSpeedCount, elem._2.normalSpeedCarCount, elem._2.lowSpeedCarCount))
  93.           }
  95.         }
  96.       })
  98.     //打印结果
  99.     top5MonitorInfoDS.addSink(new JDBCSink[Top5MonitorInfo]("Top5MonitorInfo"))
  100.     env.execute()
  101.   }
  102. }

四、实时对违法车辆-套牌车辆 进行监控

  1. import java.util.Properties
  2. import com.bjsxt.traffic.utils.{DateUtils, JDBCSink, MonitorCarInfo, ViolationCarInfo}
  3. import org.apache.flink.api.common.serialization.SimpleStringSchema
  4. import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
  5. import org.apache.flink.streaming.api.functions.KeyedProcessFunction
  6. import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
  7. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
  8. import org.apache.flink.util.Collector
  9. import org.apache.kafka.common.serialization.StringDeserializer
  11. object RepatitionCarInfo {
  12.   def main(args: Array[String]): Unit = {
  13.       val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  14.     import org.apache.flink.streaming.api.scala._
  16.     //设置kafka 配置
  17.     val props = new Properties()
  18.     props.setProperty("bootstrap.servers","mynode1:9092,mynode2:9092,mynode3:9092")
  19.     props.setProperty("key.deserializer",classOf[StringDeserializer].getName)
  20.     props.setProperty("value.deserializer",classOf[StringDeserializer].getName)
  21.     props.setProperty("group.id","group112504")
  23.     //Flink 读取Kafka 中数据
  24.     val carMonitorInfoDS: DataStream[String] = env.addSource(new FlinkKafkaConsumer[String]("monitortopic1125",new SimpleStringSchema(),props).setStartFromEarliest())
  26.     //转换数据
  27.     val transferDs: DataStream[MonitorCarInfo] = carMonitorInfoDS.map(line => {
  28.       val arr: Array[String] = line.split("\t")
  29.       MonitorCarInfo(arr(0), arr(1), arr(2), arr(3), arr(4).toLong, arr(5), arr(6).toDouble)
  30.     })
  32.     //对监控到车辆数据，使用状态编程 ，统计在10s 内通过两个卡扣的涉嫌套牌车辆
  33.     val violationCarInfoDS: DataStream[ViolationCarInfo] = transferDs.keyBy(_.car).process(new KeyedProcessFunction[String, MonitorCarInfo, ViolationCarInfo] {
  35.       //给每个车辆设置状态，状态中存储的是最近这辆车经过卡扣的时间
  36.       private lazy val carStateTime: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("carTimeState", classOf[Long]))
  38.       //value : 当前本条数据 ，ctx :上下文数据，out:回收数据对象
  39.       override def processElement(value: MonitorCarInfo, ctx: KeyedProcessFunction[String, MonitorCarInfo, ViolationCarInfo]#Context, out: Collector[ViolationCarInfo]): Unit = {
  40.         //获取当前车辆的状态值
  41.         val preLastActionTime: Long = carStateTime.value()
  42.         //判断当前车辆是否有对应的状态值
  43.         if (preLastActionTime == 0) {
  44.           //没有状态，存储状态
  45.           carStateTime.update(value.actionTime)
  46.         } else {
  47.           //获取当前数据通过卡扣拍摄时间
  48.           val actionTime: Long = value.actionTime
  49.           val car: String = value.car
  50.           //有对应的状态，就获取状态时间与当前本条数据时间做差值，看下是否超过10s
  51.           if ((actionTime - preLastActionTime).abs < 10 * 1000) {
  52.             //涉嫌套牌车辆
  53.             out.collect(new ViolationCarInfo(car,
  54.               "涉嫌套牌",
  55.               DateUtils.timestampToDataStr(System.currentTimeMillis()),
  56.               s"车辆${car}通过上一次卡扣时间${preLastActionTime},本次通过卡扣时间${actionTime},两次时间差值为${(preLastActionTime - actionTime).abs}ms"))
  57.           }
  58.           //更新状态
  59.           carStateTime.update(actionTime.max(preLastActionTime))
  60.         }
  61.       }
  62.     })
  64.     violationCarInfoDS.addSink(new JDBCSink[ViolationCarInfo]("ViolationCarInfo"))
  65.     env.execute()
  66.   }
  67. }