车辆碰撞程序

总体概述

简单介绍

车辆在发生对应的异常情况时，产生事件数据，实时程序消费通过OGG 推送至Kafka的数据进行事件判断，并将消息通过Rabbit推送至业务端。

碰撞告警：车辆发生碰撞事件并有气囊、车胎、翻滚等状态。
低等级碰撞告警：车辆发生轻微碰撞，产生低等级碰撞计数
故障告警：车辆发生故障事件

涉及组件

组件名称	说明
Kafka	消息中间件
Hbase	数据库
RabbitMq	消息中间件
flink 1.12.1	大数据计算框架

程序设计

程序需要根据监测碰撞事件发生，并获取到事件发生的位置，根据车辆号查询到对应的车架号，将结果实时发送给下游。

 * SQL 逻辑
 -- 碰撞事件位置表  VEHICLE_LOCATION
 -- 碰撞事件主表    INTERACTION
 SQL 逻辑:
     SELECT
         cast(t1.interaction_id as varchar(47)) CONTCT_ID,
         cast(t2.vehicle_vin as varchar(17))    VIN_ID,
         t3.location,
         cast(t1.aacn_data as varchar(1500))    RAW_DATA,
         t1.created_on as                       CREATE_TIMSTM
     from interaction t1
           inner join vehicle t2                               > Part II
                   on t1.vehicle_id = t2.vehicle_id             > Vehicle中查询
           left join vehicle_location t3
                   on t1.interaction_id = t3.owner_id
           where t1.interaction_type = 'crash'
           and t1.created_on >= to_date('20191226', 'yyyymmdd')
           and t1.created_on < to_date('20191227', 'yyyymmdd')

1.位置与事件信息流Join

 /**
    * Join elements of this [[KeyedStream]] with elements of another [[KeyedStream]] over
    * a time interval that can be specified with [[IntervalJoin.between]].
    *
    * @param otherStream The other keyed stream to join this keyed stream with
    * @tparam OTHER Type parameter of elements in the other stream
    * @return An instance of [[IntervalJoin]] with this keyed stream and the other keyed stream
    */
acrInterStream
      .intervalJoin(acrLocStream)
      .between(Time.minutes(-10), Time.minutes(10))
      .process(new AcrJoinProcessFunction())// Join, (vehicleId, ObjectNode)

2. 根据车辆号查询车架号

对应关系为Hbase表，通过异步算子使用Hbase Api Get出对应值。

  AsyncDataStream
        .unorderedWait(
                   acrJoinStream,
                   new AsyncAcrGetVinFromHbaseFunction(HbaseConfig),
                   3000,
                   TimeUnit.SECONDS,
                   20) //异步查询Hbase 获取vin号