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

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 '区（县）'
);

 

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

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）启动作业

@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;
}

//继承 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.返回数据

//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方法，使用集合中只放一个对象

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();
    }
}

/**
 * 实时在线故障分析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;
}

//实现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

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

/**
 * 定义车辆基础信息表的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;
}

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

//继承 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
//赋值基本配置给变量

/**
 * @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 根据位编码、经度最小值、经度最大值计算出经度
        double lon = decode(lonset, -180, 180);
        // todo 根据位编码、纬度最小值、纬度最大值计算出经度
        double lat = decode(latset, -90, 90);
        // todo 返回纬度、经度数组
        return new double[]{lat, lon};
    }

    /**
     * @desc todo 编码方法，根据编码、维度[-90,90],经度[-180,180]
     * @param bs
     * @param floor
     * @param ceiling
     * @return
     */
    private static double decode(BitSet bs, double floor, double ceiling) {
        double mid = 0;
        for (int i = 0; i < bs.length(); i++) {
            mid = (floor + ceiling) / 2;
            if (bs.get(i))
                floor = mid;
            else
                ceiling = mid;
        }
        return mid;
    }

    /**
     * @desc todo 解码方法，根据纬度、经度，返回32编码字符串
     * @param lat 纬度
     * @param lon 经度
     * @return base32的字符串
     */
    public static String encode(double lat, double lon) {
        BitSet latbits = getBits(lat, -90, 90);
        BitSet lonbits = getBits(lon, -180, 180);
        StringBuilder buffer = new StringBuilder();
        for (int i = 0; i < numbits; i++) {
            buffer.append(lonbits.get(i) ? '1' : '0');
            buffer.append(latbits.get(i) ? '1' : '0');
        }
        return base32(Long.parseLong(buffer.toString(), 2));
    }

    /**
     * @desc todo 根据经纬度和范围，获取对应的二进制值
     * @param d 经度 | 纬度
     * @param floor 最小值
     * @param ceiling 最大值
     * @return 返回BitSet，java.util工具类，用于位移操作工具类
     */
    private static BitSet getBits(double d, double floor, double ceiling) {
        BitSet buffer = new BitSet(numbits);
        for (int i = 0; i < numbits; i++) {
            double mid = (floor + ceiling) / 2;
            if (d >= mid) {
                buffer.set(i);
                floor = mid;
            } else {
                ceiling = mid;
            }
        }
        return buffer;
    }

    /**
     * @desc todo 将经纬度合并后二二进制进行指定32位编码
     * @param i 被32编码的long值
     * @return 32编码字符串
     */
    private static String base32(long i) {
        char[] buf = new char[65];
        int charPos = 64;
        boolean negative = (i < 0);
        if (!negative)
            i = -i;
        while (i <= -32) {
            buf[charPos--] = digits[(int) (-(i % 32))];
            i /= 32;
        }
        buf[charPos] = digits[(int) (-i)];

        if (negative)
            buf[--charPos] = '-';
        return new String(buf, charPos, (65 - charPos));
    }
}
/**
 * @Description TODO 测试geohashutil 验证：http://www.geohash.cn/
 */
public class TestGeoHashUtil {
    public static void main(String[] args) {
        // todo 根据经纬度编码
        String geohash = GeoHashUtil.encode(25.050583, 121.559322);
        System.out.println(geohash);
        // todo geohash值解码
        double[] geo = GeoHashUtil.decode("wsqqw0kf1x0h");
        System.out.println(geo[0]+" "+geo[1]);
    }
}

public class RedisUtil {
    private static Pool<Jedis> jedisPool = null;
    private static ReentrantLock lock = new ReentrantLock();
    private static String HOST = ConfigLoader.getProperty("redis.host");
    private static int PORT = Integer.valueOf(ConfigLoader.getProperty("redis.port"));
    private static int TIMEOUT = Integer.valueOf(ConfigLoader.getProperty("redis.session.timeout"));
    private static int DATABASE = Integer.valueOf(ConfigLoader.getProperty("redis.database"));
    private static String PASSWORD = ConfigLoader.getProperty("redis.password");

    /**
     * 初始化连接池
     */
    static {
        if ("null".equals(PASSWORD)) PASSWORD = null;
        if (jedisPool == null) {
            jedisPool = new JedisPool(new GenericObjectPoolConfig(), HOST, PORT, TIMEOUT, PASSWORD, DATABASE, "");
        }
    }

    /**
     * @desc:获得jedis客户端
     * @return Jedis客户端
     */
    public static Jedis getJedis() {
        if (jedisPool == null) {
            lock.lock(); //防止吃初始化时多线程竞争问题
            jedisPool = new JedisPool(new GenericObjectPoolConfig(), HOST, PORT, TIMEOUT, PASSWORD, DATABASE, "");
            lock.unlock();
        }
        return jedisPool.getResource();
    }

    /**
     * @desc:根据redis中存在的key获得value
     * @param key
     * @return value的字节数组
     */
    public static byte[] get(byte[] key) {
        Jedis jedis = getJedis();
        byte[] result = "".getBytes();
        if (jedis.exists(key)) {
            result = jedis.get(key);
        }
        jedis.close();
        return result;
    }

    /**
     * @desc:插入数据到redis中，并设置key的存活时间（seconds）
     * @param key
     * @param value
     * @param keyTimeout
     * @return
     */
    public static Boolean set(byte[] key, byte[] value, int keyTimeout) {
        try {
            Jedis jedis = getJedis();
            jedis.setex(key, keyTimeout, value);
            jedis.close();
            return true;
        } catch (Exception e){
            e.printStackTrace();
            return false;
        }
    }

    public static void set(byte[] key, byte[] value) {
        try {
            Jedis jedis = getJedis();
            jedis.set(key, value);
            jedis.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * @desc:释放资源，关闭连接池
     */
    public static void releaseSource() {
        if (jedisPool != null) jedisPool.close();
    }
}