实时即未来，车联网项目之远程诊断实时故障分析

Posted 码农Maynor

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了实时即未来，车联网项目之远程诊断实时故障分析相关的知识，希望对你有一定的参考价值。

文章目录

远程诊断实时故障业务

什么是远程诊断实时故障

监管部门或者车企通过判断实时上报的车辆数据，从而研判当前车辆故障诊断信息，给驾驶员发送预警告警信息等。

应用场景介绍

① 内部管理系统针对车辆的故障查询统计信息

② 实时监控大屏

常用故障分析指标与含义

  • 19项车辆故障指标和车辆报警、故障信息属性50+

  • *报警指标**报警指标内容**值与含义*
    batteryAlarm电池高温报警0：正常1：异常
    singleBatteryOverVoltageAlarm单体电池高压报警
    batteryConsistencyDifferenceAlarm电池单体一致性差报警
    insulationAlarm绝缘报警
    highVoltageInterlockStateAlarm高压互锁状态报警
    socJumpAlarmSOC跳变报警
    driveMotorControllerTemperatureAlar驱动电机控制器温度报警
    dcdcTemperatureAlarmDC-DC温度报警（dc-dc可以理解为车辆动力智能系统转换器）
    socHighAlarmSOC过高报警
    socLowAlarmSOC低报警
    temperatureDifferenceAlarm温度差异报警
    vehicleStorageDeviceUndervoltageAlarm车载储能装置欠压报警
    dcdcStatusAlarmDC-DC状态报警
    singleBatteryUnderVoltageAlarm单体电池欠压报警
    rechargeableStorageDeviceMismatchAlarm可充电储能系统不匹配报警
    vehicleStorageDeviceOvervoltageAlarm车载储能装置过压报警
    brakeSystemAlarm制动系统报警
    driveMotorTemperatureAlarm驱动电机温度报警
    vehiclePureDeviceTypeOvercharge车载储能装置类型过充报警

业务中间表数据结构

  • 涉及到8张表和1章分析结果表

    [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-5GOAHsGl-1664278460628)(assets/image-20210927092525648.png)]

  • 表字段介绍

    分析的结果表 online_data 分为三类：

    1. 实时上报的车辆数据；
    2. 静态的车辆车型车系等数据；
    3. 通过经纬度获取到的国家、省市区、地址等信息。

分析结果表数据结构

create table online_data
(
    vin                 varchar(17)   not null comment '车架号' primary key,
    process_time        datetime      null comment '数据更新时间',
    lat                   double        null comment '纬度',
    lng                   double        null comment '经度',
    mileage               double        null comment '里程表读数',
    is_alarm              int(1)        null comment '故障标志(0正常，1故障)',
    alarm_name            varchar(1000) null comment '故障名称(多个故障用~分割)',
    terminal_time         datetime      null comment '终端时间',
    earliest_time         datetime      null comment '最早数据接收时间',
    max_voltage_battery   double        null comment '单体电池最高电压',
    min_voltage_battery   double        null comment '单体电池最低电压',
    max_temperature_value double        null comment '电池最高温度',
    min_temperature_value double        null comment '电池最低温度',
    speed                 double        null comment '车速',
    soc                   int(3)        null comment 'SOC',
    charge_flag           int(1)        null comment '充电标识 0：未充电 1：充电 2：异常',
    total_voltage         double        null comment '总电压,单位：V，实际取值0.1~100V',
    total_current         double        null comment '总电流,单位：A,实际取值为-1000~1000A',
    battery_voltage       varchar(1000) null comment '单体电池电压列表',
    probe_temperatures    varchar(1000) null comment '电池模块温度列表',
    series_name           varchar(255)  null comment '车系',
    model_name            varchar(255)  null comment '车型',
    live_time             int           null comment '年限(单位:月，未查到数据显示-1)',
    sales_date            varchar(20)   null comment '销售日期',
    car_type              varchar(20)   null comment '车辆类型',
    province              varchar(255)  null comment '省份',
    city                  varchar(255)  null comment '城市',
    county                varchar(20)   null comment '区（县）'
);

高德地图解决逆地理坐标问题

  • 拟地理编码含义

    输入位置信息（经度和维度）获取地球上位置。

  • 高德等第三方Api支持拟地理演示地址

    https://developer.amap.com/demo/javascript-api/example/geocoder/regeocoding/

  • 如果使用高德Api的步骤

    1. 获取key

    2. 将key和经纬度参数封装为 url

    3. 异步请求 httpGet 获取位置数据

    4. 返回位置数据

  • 远程实时诊断地理位置查询实现思路

实时故障分析任务

  • 分析任务流程分析步骤

    1. 消费数据，转换json对象
    2. 过滤数据
    3. 根据vin分组，创建timeWindow
    4. 自定义window funtion，设置输出对象
    5. 加载车型、车型、销售等数据并广播
    6. 窗口数据与广播数据连接
    7. 获得地理位置信息数据并与窗口数据连接
    8. 结果落地数据到mysql中

  • 实时故障分析流程

远程诊断实时故障分析

  • 创建远程诊断实时故障分析任务主类—— OnlineStatisticsTask

  • 开发步骤

    1）初始化flink流处理的运行环境（事件时间、checkpoint、hadoop name）
2）接入kafka数据源，消费kafka数据
3）将消费到的json字符串转换成ItcastDataPartObj对象
4）过滤掉异常数据，保留正常数据
5）与redis维度表进行关联拉宽地理位置信息，对拉宽后的流数据关联redis，根据geohash找到地理位置信息，进行拉宽操作
6）过滤出来redis拉宽成功的地理位置数据
7）过滤出来redis拉宽失败的地理位置数据
8）对redis拉宽失败的地理位置数据使用异步io访问高德地图逆地理位置查询地理位置信息，并将返回结果写入到redis中
9）将reids拉宽的地理位置数据与高德api拉宽的地理位置数据进行合并
10）创建原始数据的30s的滚动窗口，根据vin进行分流操作
11）对原始数据的窗口流数据进行实时故障分析（区分出来告警数据和非告警数据19个告警字段）
12）加载业务中间表（7张表：车辆表、车辆类型表、车辆销售记录表，车俩用途表4张），并进行广播
13）将第11步和第12步的广播流结果进行关联，并应用拉宽操作
14）将拉宽后的结果数据写入到mysql数据库中
15）启动作业

  • 需要获取地理位置对象，可以作为ItcastDataPartObj的父类

    @Data
@NoArgsConstructor
@AllArgsConstructor
public class VehicleLocationModel implements Serializable 
    //省份
    private String province;
    //城市
    private String city;
    //国家
    private String country;
    //区县
    private String district;
    //详细地址
    private String address;
    //纬度
    private Double lat = -999999D;
    //经度
    private Double lng = -999999D;

  • 实现合并流数据在redis存储的地理位置数据拉宽操作——LocationInfoRedisFunction

    //继承 RichMapFunction<ItcastDataPartObj, ItcastDataPartObj>
//1.重写 map 方法
//1.1.获取车辆数据的经度和维度生成 geohash
//1.2.根据geohash 从redis中获取value值（geohash在redis中是作为主键存在）
//1.3.如果查询出来的值不为空，将其通过JSON对象转换成 VehicleLocationModel 对象，否则置为 null
//1.4.如果当前对象不为空，将国家，省市区地址赋值给 itcastDataPartObj，否则置为 null
//1.5.返回数据

  • 对在redis获取失败的经纬度使用异步io流请求高德Api——AsyncHttpQueryFunction

    //1.重写open方法
//1.1.创建请求配置
//1.2.创建Http异步的客户端
//1.3.开启client
//2.重写close方法
//3.重写timeout方法
//3.1.打印输出超时
//4.重写asyncInvoke方法
//4.1.获取当前车辆的经纬度
//4.2.通过GaoDeMapUtils工具类根据参数获取请求的url
//4.3.创建 http get请求对象
//4.4.使用刚创建的http异步客户端执行 http请求对象
//4.5.从执行完成的future中获取数据，返回ItcastDataPartObj对象
//4.5.1.重写get方法
//4.5.1.1.使用future获取到返回的值
//判断如果返回值的状态是正常值 200
//获取到响应的实体对象 entity
//将实体对象使用EntityUtils转换成string字符串
//因为返回的是json，需要使用JSON转换成JSONObject对象
//通过regeocode获取JSON对象，然后解析对象封装国家，省市区，地址
//封装成 VehicleLocationModel 对象
//4.5.1.2.通过RedisUtil将数据写入到redis，
//key=geohash，value=封装的对象的JSON字符串toJSONString
//4.5.1.3.将国家，省市区，地址进行封装并返回
//4.6.从future的thenAccept
//4.6.1.重写accept方法，使用集合中只放一个对象

  • 引入高德Api 访问的工具类

    public class GaoDeMapUtils 
    //指定高德地图请求的密钥
    private static final String KEY = ConfigLoader.getProperty("gaode.key");
    //指定返回值类型
    private static final String OUTPUT = "json";
    //请求的地址
    private static final String GET_ADDRESS_URL = ConfigLoader.getProperty("gaode.address.url");

    /**
     * 传递经纬度返回逆地理位置查询的请求地址
     * @param longitude
     * @param latitude
     * @return
     */
    public static String getUrlByLonLat(double longitude, double latitude) 
        //拼接经纬度的字符串参数
        String location = longitude + "," + latitude;
        //定义参数的集合对象
        Map<String, String> params = new HashMap<>();
        params.put("location", location);

        //根据请求base地址和参数集合列表拼接出来请求的完整地址
        String url = joinUrl(params, GET_ADDRESS_URL);
        return url;
    

    /**
     * 拼接请求的参数和请求地址
     * @param params
     */
    private static String joinUrl(Map<String, String> params, String url) 
        StringBuilder baseUrl = new StringBuilder();
        baseUrl.append(url);

        try 
            //指定参数的索引
            int index = 0;
            Set<Map.Entry<String, String>> entries = params.entrySet();
            for (Map.Entry<String, String> param : entries) 
                if (index == 0) 
                    baseUrl.append("?");
                 else 
                    baseUrl.append("&");
                
                //拼接所有的参数
                baseUrl.append(param.getKey()).append("=").append(URLEncoder.encode(param.getValue(), "utf-8"));
            
            baseUrl.append("&output=").append(OUTPUT).append("&key=").append(KEY);
         catch (UnsupportedEncodingException e) 
            e.printStackTrace();
        
        return baseUrl.toString();

  • 导入实时在线故障分析对象，用于存储在线故障对象——OnlineDataObj

    /**
 * 实时在线故障分析javaBean对象
 */
@Data
public class OnlineDataObj extends VehicleLocationModel  
    //车架号
    private String vin;
    //数据更新时间
    private String processTime;
    //里程表读数
    private double mileage;
    //故障标志(0正常，1故障)
    private int isAlarm;
    //故障名称(多个故障用~分割)
    private String alarmName;
    //终端时间
    private String terminalTime;
    //最早数据接收时间
    private String earliestTime;
    //单体电池最高电压
    private double maxVoltageBattery;
    //单体电池最低电压
    private double minVoltageBattery;
    //电池最高温度
    private double maxTemperatureValue;
    //电池最低温度
    private double minTemperatureValue;
    //车速
    private double speed;
    //SOC
    private int soc;
    //充电标识 0：未充电 1：充电 2：异常
    private int chargeFlag;
    //总电压,单位：V，实际取值0.1~100V
    private double totalVoltage;
    //总电流,单位：A,实际取值为-1000~1000A
    private double totalCurrent;
    //单体电池电压列表
    private String batteryVoltage;
    //电池模块温度列表
    private String probeTemperatures;
    //车系
    private String seriesName;
    //车型
    private String modelName;
    //年限(单位:月，未查到数据显示-1)
    private String liveTime;
    //销售日期
    private String salesDate;
    //车辆类型
    private String carType;
    //省份
    private String province;
    //城市
    private String city;
    //国家
    private String county;
    //区县
    private String district;
    //详细地址
    private String address;

  • 窗口自定义远程故障诊断自定义窗口实现——OnlineStatisticsWindowFunction

    //实现WindowFunction<ItcastDataPartObj, OnlineDataObj, String, TimeWindow>接口
//1.对当前的数据集合进行升序排列
//2.获取集合中第一条数据
//3.循环遍历每条数据，将集合中存在异常的数据拼接到指定属性中
//30s窗口最多6条数据，每条数据需要检测19个字段，如果出现异常字段就进行  //字符串拼接
//3.1.过滤没有各种告警的信息，调用setOnlineDataObj 将第一条对象和每条对象和标识0 返回到OnlineDataObj，并收集这个对象
// 否则 调用setOnlineDataObj 将第一条对象和每条对象和标识1 返回到OnlineDataObj，并收集这个对象
//4.实现setOnlineDataObj 方法
//4.1.定义OnlineDataObj
//4.2.将每条的对象属性拷贝到定义OnlineDataObj
//4.3.将每条对象中表显里程赋值给mileage
//4.4.将告警信号赋值给isAlarm
//4.5.将每个对象通过addAlarmNameList生成告警list，拼接成字符串赋值给alarmName，通过字符串join
//4.6.将窗口内第一条数据告警时间赋值给 earliestTime
//4.7.将获取每条记录的充电状态通过getChargeState返回充电标识赋值给充电标记
//4.8.将当前时间赋值给处理时间
//引入-判断是否存在报警的字段，addAlarmNameList，getChargeState

车型车系销售信息广播流

  • 涉及到的字段

    车型、车系、车辆销售信息数据，主要获得9个字段信息：

    vin、series_name、model_name、series_code、model_code、nick_name、sales_date、product_date、car_type

  • 数据源模型

  • 从MySQL中读取车型车系销售信息

    select t12.vin,t12.series_name,t12.model_name,t12.series_code,t12.model_code,t12.nick_name,t3.sales_date product_date,t4.car_type
 from (select t1.vin, t1.series_name, t2.show_name as model_name, t1.series_code,t2.model_code,t2.nick_name,t1.vehicle_id
 from vehicle_networking.dcs_vehicles t1 left join vehicle_networking.t_car_type_code t2 on t1.model_code = t2.model_code) t12
 left join  (select vehicle_id, max(sales_date) sales_date from vehicle_networking.dcs_sales group by vehicle_id) t3
 on t12.vehicle_id = t3.vehicle_id
 left join
 (select tc.vin,'net_cat' car_type from vehicle_networking.t_net_car tc
 union all select tt.vin,'taxi' car_type from vehicle_networking.t_taxi tt
 union all select tp.vin,'private_car' car_type from vehicle_networking.t_private_car tp
 union all select tm.vin,'model_car' car_type from vehicle_networking.t_model_car tm) t4
 on t12.vin = t4.vin

  • [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-eqD9pRLV-1664278460629)(assets/image-20210927174106033.png)]

  • 导入车辆车型车系结果对象

    /**
 * 定义车辆基础信息表的javaBean对象
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
public class VehicleInfoModel 
    //车架号
    private String vin;
    //车型编码
    private String modelCode;
    //车型名称
    private String modelName;
    //车系编码
    private String seriesCode;
    //车系名称
    private String seriesName;
    //出售日期
    private String salesDate;
    //车型
    private String carType;
    //车辆类型简称
    private String nickName;
    //年限
    private String liveTime;

  • 创建读取MySQL的Flink的数据源Source——VehicleInfoMysqlSource

    • 实现RichSourceFunction<HashMap<String, VehicleInfoModel>>

    • 将数据源广播出去

    //自定义实现车辆基础信息表的加载
//加载车辆基础信息表（车辆类型、车辆、销售记录表、车辆用途表）
//重写open方法
//重写run方法
//重写close方法
//重写cancel方法

  • 窗口流数据与广播数据connect再flatMap——VehicleInfoMapMysqlFunction

    //继承 RichCoFlatMapFunction<OnlineDataObj, HashMap<String, VehicleInfoModel>, OnlineDataObj>
//1.重写flatMap1
//1.1.通过 vin 获取到车辆基础信息
//1.2.如果车辆基础信息不为空
//1.2.1.将车系seriesName，车型modelName,年限LiveTime,销售日期saleDate，车辆类型carType封装到onlineDataObj对象中
//1.2.2.将onlineDataObj收集返回
//1.3.打印输出，基础信息不存在
//2.重写flatMap2
//赋值基本配置给变量

获取地理位置信息

基于geohash编码的地理位置计算

  • geohash的概念介绍（高效的多维空间点索引算法.html）

    geohash 就是将地图上位置（经纬度）转换成偶数位是经度、奇数数是维度，新的二进制字节，转换成字符串，用字符串代表某一个地理位置。

  • geohash编码实现

    /**
 * @Description TODO geohash算法实现工具类
 */
public class GeoHashUtil 
    // todo 经纬度编码长度
    private static int numbits =  6 * 5;
    // todo 数字字母编码数组
    final static char[] digits = '0', '1', '2', '3', '4', '5', '6', '7',
            '8', '9', 'b', 'c', 'd', 'e', 'f', 'g',
            'h', 'j', 'k', 'm', 'n', 'p', 'q', 'r',
            's', 't', 'u', 'v', 'w', 'x', 'y', 'z';
    // todo 存储编码字符循环存储的HashMap对象
    final static HashMap<Character, Integer> lookup = new HashMap<>();

    // todo 静态代码块，执行设置HashMap的key，value
    static 
        int i = 0;
        for (char c : digits)
            lookup.put(c, i++);
    

    /**
     * @desc todo 根据编码后的geohash字符串值，进行解码，得到经纬度数组
     * @param geoHash
     * @return [lat, lon]
     */
    public static double[] decode(String geoHash) 
        StringBuilder buffer = new StringBuilder();
        for (char c : geoHash.toCharArray()) 
            int i = lookup.get(c) + 32;
            buffer.append(Integer.toString(i, 2).substring(1));
        

        BitSet lonset = new BitSet();
        BitSet latset = new BitSet();

        // todo 经度，偶数位
        int j = 0;
        for (int i = 0; i < numbits * 2; i += 2) 
            boolean isSet = false;
            if (i < buffer.length())
                isSet = buffer.charAt(i) == '1';
            lonset.set(j++, isSet);
        

        // todo 纬度，奇数位
        j = 0;
        for (int i = 1; i < numbits * 2; i += 2) 
            boolean isSet = false;
            if (i < buffer.length())
                isSet = buffer.charAt(i) == '1';
            latset.set(j++, isSet);
        

        // todo 根据位编码、经度最小值
    以上是关于实时即未来，车联网项目之远程诊断实时故障分析的主要内容，如果未能解决你的问题，请参考以下文章
     

     实时即未来，车联网项目之远程诊断实时故障分析
    

     实时即未来，大数据项目车联网之车辆驾驶行程分析十三
    

     实时即未来，大数据项目车联网之车辆驾驶行程分析十三
    

     实时即未来，大数据项目车联网之车联网之车辆数据统计意义分析十三
    

     实时即未来，车联网项目之车辆驾驶行为分析
    

     实时即未来，车联网项目之车辆驾驶行为分析