实时即未来，车联网项目之电子围栏分析【六】 |
您所在的位置：网站首页 › 电子围栏介绍怎么写范文图片大全 › 实时即未来，车联网项目之电子围栏分析【六】 |
实时即未来，车联网项目之电子围栏分析【六】
|
文章目录
电子围栏简介和应用场景电子围栏规则和分析结果数据结构电子围栏分析步骤电子围栏分析任务实现广播状态与实现电子围栏中的 ConnectStreamed应用connect流说明connect流使用场景两点之间球面距离的计算——DistanceCaculateUtil电子围栏中自定义对象将两个数据流合并
设置窗口并计算确定是否在电子围栏内告警合并分析电子围栏结果读取电子围栏分析结果并广播窗口流数据与广播流数据连接
电子围栏分析结果入库测试电子围栏
电子围栏简介和应用场景
电子围栏简介和意义
地理围栏是一个虚拟的空间围栏，可以帮助开发者检测人或物何时进入或离开预定义区域，并支持实时报警功能。 电子围栏的应用场景 签到打卡类场景共享单车类场景线下门店促销场景 创建电子围栏 在此项目中，使用的电子围栏是规则的圆形，判断是否在圆形电子围栏区域内，可以使用车辆位置和中心点球面距离小于等于半径，在电子围栏的区域内。还有一些不规则的电子围栏，这些可以使用射线取点的个数来判断是否在电子围栏内，如果是偶数在电子围栏外，否则是电子围栏内。 电子围栏规则和分析结果数据结构电子围栏的定义
电子围栏规则数据结构
数据样本示例
电子围栏分析结果数据结构
电子围栏分析任务设置、原始数据json解析、过滤异常数据 读取已存在电子围栏中的车辆与电子围栏信息(广播流临时结果数据) 原始车辆数据与电子围栏广播流进行合并，生成电子围栏规则模型流数据（DStream） 创建90秒翻滚窗口，计算电子围栏信息(ElectricFenceModel中的值根据车辆是否在围栏内进行设置) 读取电子围栏分析结果表数据并广播 翻滚窗口电子围栏对象模型流数据与电子围栏分析结果数据广播流进行connect 对电子围栏对象模型，添加uuid和inMySQL(车辆是否已存在mysql表中) 电子围栏分析结果数据落地mysql，也可以选择落地mongo 电子围栏分析任务实现电子栅栏分析的逻辑图 电子围栏分析主类：ElectricFenceTask 简化 ItcastDataObj 对象：ItcastDataPartObj.java 简化解析 ItcastParseUtil 对象： JsonParsePartUtil.java 测试工具类对象 //实现步骤: 1）初始化flink流处理的运行环境（设置按照事件时间处理数据、设置hadoopHome的用户名、设置checkpoint） 2）读取kafka数据源（调用父类的方法） 3）将字符串转换成javaBean（ItcastDataPartObj）对象 4）过滤出来正常数据 5）读取电子围栏规则数据以及电子围栏规则关联的车辆数据并进行广播 6）将原始数据（消费的kafka数据）与电子围栏规则数据进行关联操作（Connect）并flatMap为 ElectricFenceRulesFuntion 7）对上步数据分配水印（30s）并根据 vin 分组后应用90s滚动窗口，然后对窗口进行自定义函数的开发（计算出来该窗口的数据属于电子围栏外还是电子围栏内） 8）读取电子围栏分析结果表的数据并进行广播 9）对第七步和第八步产生的数据进行关联操作（connect） 10）对第九步的结果进行滚动窗口操作，应用自定义窗口函数（实现添加uuid和inMysql属性赋值） 11）将分析后的电子围栏结果数据实时写入到mysql数据库中 12）运行作业，等待停止 广播状态与实现 回顾广播变量概念广播变量就是将变量广播到各个 taskmanager的内存中，可以共享数据，一般情况下广播变量的类型是 map 类型 key->value 广播变量的数据格式是——map类型state 如何使用广播变量 HashMap ，其中String:vin 电子围栏转换临时对象——ElectricFenceResultTmp @Data @AllArgsConstructor public class ElectricFenceResultTmp { //电子围栏id private int id; //电子围栏名称 private String name; //电子围栏中心地址 private String address; //电子围栏半径 private float radius; //电子围栏中心点的经度 private double longitude; //电子围栏中心点的维度 private double latitude; //电子围栏的开始时间 private Date startTime; //电子围栏的结束时间 private Date endTime; @Override public String toString() { return "ElectricFenceResultTmp{" + "id=" + id + ", name='" + name + '\'' + ", address='" + address + '\'' + ", radius=" + radius + ", longitude=" + longitude + ", latitude=" + latitude + ", startTime=" + startTime + ", endTime=" + endTime + '}'; } }自定义 source 读取 MySQL 的数据源并广播 定义读取电子围栏规则类——MysqlElectricFenceSouce 返回类型为 HashMap //读取mysql存储的电子围栏规则表数据以及电子围栏规则关联的电子围栏规则车辆表数据，根据分析，一个车辆可能适配多个电子围栏规则，所以返回的数据类型定义为HashMap，为了方便处理，我们只处理一个车辆关联一个电子围栏规则的场景（真事的业务开发中一定是一个车辆可能有很多很多对应电子围栏规则的）。 //继承 RichSourceFunction //1.重写 open 方法 //1.1 获取上下文中的 parameterTool //1.2 读取配置文件中，注册驱动 url user password //1.3 实例化statement //2.重写 close 方法 //2.1 关闭 statement 和 conn //3.重写 run 方法 //3.1 每指定时间循环读取 mysql 中的电子围栏规则 //3.2 收集 electricFenceResult 指定休眠时间 //4.重写 cancel 方法读取数据库中配置信息 select vins.vin,setting.id,setting.name,setting.address,setting.radius,setting.longitude,setting.latitude,setting.start_time,setting.end_time from vehicle_networking.electronic_fence_setting setting inner join vehicle_networking.electronic_fence_vins vins on setting.id=vins.setting_id where setting.status=1 电子围栏中的 ConnectStreamed应用 connect流说明 connect流使用场景 两点之间球面距离的计算——DistanceCaculateUtil导入工具jar包坐标 org.gavaghan geodesy ${geodesy.version}两点之间球面距离的计算工具类 /** * TODO 球面距离计算工具类;根据两个点的经纬度，计算出距离 */ public class DistanceCaculateUtil { /** * @desc:计算地址位置方法，坐标系、经纬度用于计算距离(直线距离) * @param gpsFrom * @param gpsTo * @param ellipsoid * @return 计算距离 */ private static Double getDistanceMeter(GlobalCoordinates gpsFrom, GlobalCoordinates gpsTo, Ellipsoid ellipsoid) { // GeodeticCurve geodeticCurve = new GeodeticCalculator().calculateGeodeticCurve(ellipsoid, gpsFrom, gpsTo); return geodeticCurve.getEllipsoidalDistance(); } /** * @desc:使用传入的ellipsoidsphere方法计算距离 * @param latitude 位置1经度 * @param longitude 位置1维度 * @param latitude2 位置2经度 * @param longitude2 位置2维度 * @param ellipsoid 椭圆计算算法 * @return */ private static Double ellipsoidMethodDistance(Double latitude, Double longitude, Double latitude2, Double longitude2, Ellipsoid ellipsoid){ // todo 位置点经度、维度不为空 位置点2经度、维度不为空 椭圆算法 Objects.requireNonNull(latitude, "latitude is not null"); Objects.requireNonNull(longitude, "longitude is not null"); Objects.requireNonNull(latitude2, "latitude2 is not null"); Objects.requireNonNull(longitude2, "longitude2 is not null"); Objects.requireNonNull(ellipsoid, "ellipsoid method is not null"); // todo 地球坐标对象：封装经度维度坐标对象 GlobalCoordinates source = new GlobalCoordinates(latitude, longitude); GlobalCoordinates target = new GlobalCoordinates(latitude2, longitude2); // todo 椭圆范围计算方法 return getDistanceMeter(source, target, ellipsoid); } /** * @desc:使用ellipsoidsphere方法计算距离 * @param latitude * @param longitude * @param latitude2 * @param longitude2 * @return distance 单位：m */ public static Double getDistance(Double latitude,Double longitude,Double latitude2,Double longitude2) { // 椭圆范围计算方法：Ellipsoid.Sphere return ellipsoidMethodDistance(latitude, longitude, latitude2, longitude2, Ellipsoid.Sphere); } } 电子围栏中自定义对象将两个数据流合并通过关联两个数据流后CoFlatMap 后生成实体类—— ElectricFenceModel /** * 电子围栏规则计算模型 */ @Data @AllArgsConstructor @NoArgsConstructor public class ElectricFenceModel implements Comparable { //车架号 private String vin = ""; //电子围栏结果表UUID private Long uuid = -999999L; //上次状态 0 里面 1 外面 private int lastStatus = -999999; //当前状态 0 里面 1 外面 private int nowStatus = -999999; //位置时间 yyyy-MM-dd HH:mm:ss private String gpsTime = ""; //位置纬度-- private Double lat = -999999D; //位置经度-- private Double lng = -999999D; //电子围栏ID private int eleId = -999999; //电子围栏名称 private String eleName = ""; //中心点地址 private String address = ""; //中心点纬度 private Double latitude; //中心点经度 private Double longitude = -999999D; //电子围栏半径 private Float radius = -999999F; //出围栏时间 private String outEleTime = null; //进围栏时间 private String inEleTime = null; //是否在mysql结果表中 private Boolean inMysql = false; //状态报警 0：出围栏 1：进围栏 private int statusAlarm = -999999; //报警信息 private String statusAlarmMsg = ""; //终端时间 private String terminalTime = ""; // 扩展字段 终端时间 private Long terminalTimestamp = -999999L; @Override public int compareTo(ElectricFenceModel o) { if(this.getTerminalTimestamp() > o.getTerminalTimestamp()){ return 1; } else if(this.getTerminalTimestamp() return 0; } } }实现将两个流合并CoFlatMapFunction接口—— ElectricFenceRulesFuntion //1.定义返回的 ElectricFenceModel //2.判断如果流数据数据质量（车辆的经纬度不能为0或-999999，车辆GpsTime不能为空） //2.1.获取当前车辆的 vin //2.2.通过vin获取电子围栏的配置信息 //2.3.如果电子围栏配置信息不为空 //2.3.1.说明当前车辆关联了电子围栏规则，需要判断当前上报的数据是否在电子围栏规则的生效时间内，先获取上报地理位置时间gpsTimestamp //2.3.2.如果当前gpsTimestamp>=开始时间戳并且gpsTimestamp |
【本文地址】
公司简介
联系我们
今日新闻 |
点击排行 |
|
推荐新闻 |
图片新闻 |
|
专题文章 |