[大数据]MapReduce对AIS数据进行数据插值

对数据进行插值

对于某些距离偏差过大的数据之间进行插值

首先需要掌握的几个小知识

在插值之前，首先应该对数据已经按照了某种规则完成了排序，然后按照规则进行相邻点之间的比较，判断是否符合设定的阈值，满足插值情况的就对数据进行插值

大体流程：

遍历values，然后将value逐一存放到list中，随后遍历list（存放在list中，可以比较相邻点），判断在时间差距的x是否超过了规定的距离阈值，如果超过则插值，并将数据放到list中，如果不超限就直接输出放到list中，如果对于时间超限但是距离并未超限的，可以看做是停靠点，可以舍弃。

在以上的流程中，需要了解在WGS84坐标系中两点之间距离的计算

可以使用第三方的jar包geodesy

导入依赖

<!--引入坐标求得距离的依赖-->
<dependency>
    <groupId>org.gavaghan</groupId>
    <artifactId>geodesy</artifactId>
    <version>1.1.3</version>
</dependency>

求解距离的方法

/**
     * 求解两个坐标之间的距离
     *
     * @param gpsFrom   第一个点坐标
     * @param gpsTo     第二个点坐标
     * @param ellipsoid 使用的坐标系
     * @return
     */
    public static double getDistanceMeter(GlobalCoordinates gpsFrom, GlobalCoordinates gpsTo, Ellipsoid ellipsoid) {
        //创建GeodeticCalculator，调用计算方法，传入坐标系、经纬度用于计算距离
        GeodeticCurve geoCurve = new GeodeticCalculator().calculateGeodeticCurve(ellipsoid, gpsFrom, gpsTo);

        return geoCurve.getEllipsoidalDistance();
    }


    /**
     * 求得距离
     */
    public double getStance(AISInterBean ais1, AISInterBean ais2) {

        GlobalCoordinates source = new GlobalCoordinates(ais1.getLat_d(), ais2.getLon_d());
        GlobalCoordinates target = new GlobalCoordinates(ais2.getLat_d(), ais2.getLon_d());
        
        double meter = getDistanceMeter(source, target, Ellipsoid.WGS84);

        return meter;
    }

除此之外还需要了解的细节：

因为在计算中，会出现大量的数据类型的转换，比如整型转换为double类型等，以及一些数学运算，比如取平均值，绝对值等

//java中取绝对值 Math.abs()
if (Math.abs(aisList.get(i + 1).getUnixTime() - aisList.get(i).getUnixTime()) > STANDTIME)

Math.pow(x,y)  x的y次方
        double lat = Math.pow((ais1.getLat_d() - ais2.getLat_d()),2);
        double lon = Math.pow((ais1.getLon_d() - ais2.getLon_d()),2);
        //平方根就是1/2次方
        double stance = Math.pow((lat + lon),0.5);

//Math.round():Double类型转换为Long
//long转换为double,(double)long;
Long unixTime = Math.round(ais1.getUnixTime() + (ais2.getUnixTime() - ais1.getUnixTime()) * i / (number + 1));

基本逻辑捋清楚之后就可以编写程序，在Bean、Mapper和Driver阶段都是相似的，主要是在Reducer阶段的处理

Reducer阶段代码：

package com.gis507.test.interpolation;

import com.gis507.test.AISDataSort.AISDataBean;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.gavaghan.geodesy.Ellipsoid;
import org.gavaghan.geodesy.GeodeticCalculator;
import org.gavaghan.geodesy.GeodeticCurve;
import org.gavaghan.geodesy.GlobalCoordinates;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

@Slf4j
public class AISInterReducer extends Reducer<Text,AISInterBean,Text,AISInterBean> {

    //设定的最大时间
    private static final long STANDTIME = 10;
    //设定的最大距离
    private static final double STANDSTANCE = 20;

    @Override
    protected void reduce(Text key, Iterable<AISInterBean> values, Context context) throws IOException, InterruptedException {

        //实例化一个列表，用于存放数据
        List<AISInterBean> aisList = new ArrayList<>(1000);

        //遍历values
        for (AISInterBean value : values) {

            //将value添加到列表
            aisList.add(new AISInterBean(value.getSpeed(), value.getUnixTime(), value.getLon_d(), value.getLat_d(),
                    value.getDraught(), value.getLength(), value.getWidth(), value.getCallsign(), value.getCourse(),
                    value.getRot()));

        }

        log.info("aisList的长度是：" + aisList.size());

        //所有插值得到的数据放到一个列表中
        ArrayList<AISInterBean> allAisInterList = new ArrayList<>();

        //所有满足的数据
        ArrayList<AISInterBean> rightAisList = new ArrayList<>();

        //aisList是一个放有数据的列表
        for (int i = 0; i < aisList.size() - 1; i++) {

            //判断两点之间的差是否大于规定的最大时间差
            //java中取绝对值 Math.abs()
            if (Math.abs(aisList.get(i + 1).getUnixTime() - aisList.get(i).getUnixTime()) > STANDTIME) {

                //判断是否距离超限，如果超限就插值，不超限就说明可能是停靠，就可以分段
                if (getStance(aisList.get(i), aisList.get(i + 1)) > STANDSTANCE) {

                    //往数据里插值
                    ArrayList<AISInterBean> aisInterList = insertPoint(aisList.get(i), aisList.get(i + 1));
                    //将本次插值的数据放到总的插值列表中
                    allAisInterList.addAll(aisInterList);
                } else {
                    log.info("这个点应该是个停靠点：" + aisList.get(i));
                }

            } else {
                //满足的数据列表
                rightAisList.add(new AISInterBean(aisList.get(i).getSpeed(), aisList.get(i).getUnixTime(), aisList.get(i).getLon_d(),
                        aisList.get(i).getLat_d(), aisList.get(i).getDraught(), aisList.get(i).getLength(), aisList.get(i).getWidth(),
                        aisList.get(i).getCallsign(), aisList.get(i).getCourse(), aisList.get(i).getRot()));
                log.info("符合条件的添加成功一条数据");
            }
        }

        //汇总数据
        rightAisList.addAll(allAisInterList);

        //写出
        for (AISInterBean aisInterBean : rightAisList) {

            context.write(key, aisInterBean);
        }


    }

    /**
     * 求解两个坐标之间的距离
     *
     * @param gpsFrom   第一个点坐标
     * @param gpsTo     第二个点坐标
     * @param ellipsoid 使用的坐标系
     * @return
     */
    public static double getDistanceMeter(GlobalCoordinates gpsFrom, GlobalCoordinates gpsTo, Ellipsoid ellipsoid) {
        //创建GeodeticCalculator，调用计算方法，传入坐标系、经纬度用于计算距离
        GeodeticCurve geoCurve = new GeodeticCalculator().calculateGeodeticCurve(ellipsoid, gpsFrom, gpsTo);

        return geoCurve.getEllipsoidalDistance();
    }


    /**
     * 求得距离
     */
    public double getStance(AISInterBean ais1, AISInterBean ais2) {


        GlobalCoordinates source = new GlobalCoordinates(ais1.getLat_d(), ais2.getLon_d());
        GlobalCoordinates target = new GlobalCoordinates(ais2.getLat_d(), ais2.getLon_d());


        double meter = getDistanceMeter(source, target, Ellipsoid.WGS84);

        return meter;
    }


    /**
     * 插值
     */
    public ArrayList<AISInterBean> insertPoint(AISInterBean ais1, AISInterBean ais2) {

        //获得实际距离
        double stance = getStance(ais1, ais2);
        //按照距离进行插值，需要插入点的个数
        double number = Math.floor(stance / STANDSTANCE) - 1;

        //实例化一个列表，将所有添加的数据放大集合中
        ArrayList<AISInterBean> aisList = new ArrayList<>();

        //插入的点放入集合中
        for (int i = 1; i <= number; i++) {

            //speed就应该是大于ais1，并且小于ais2
            double speed = ais1.getSpeed() + (ais2.getSpeed() - ais1.getSpeed()) * i / (number + 1);

            //插入点的lat和lon
            double lat = ais1.getLat_d() + (ais2.getLat_d() - ais1.getLat_d()) * i / (number + 1);
            double lon = ais1.getLon_d() + (ais2.getLon_d() - ais1.getLon_d()) * i / (number + 1);

            //时间
            //Math.round():Double类型转换为Long
            //long转换为double,(double)long;
            Long unixTime = Math.round(ais1.getUnixTime() + (ais2.getUnixTime() - ais1.getUnixTime()) * i / (number + 1));

            //course,rot等其他字段，都采用ais1的
            //speed,unixTime,lon_d,lat_d,draught,length,width,callsign,course,rot
            String draught = ais1.getDraught();
            String length = ais1.getLength();
            String width = ais1.getWidth();
            String callsign = ais1.getCallsign();
            double course = ais1.getCourse();
            double rot = ais1.getRot();

            //封装到aisData中
            AISInterBean aisData = new AISInterBean(speed, unixTime, lon, lat, draught, length, width, callsign, course, rot);

            //输出插值得到的点
            log.info("插值得到的点：" + aisData);
            //放到列表中
            aisList.add(aisData);
        }
        //返回列表
        return aisList;
    }
}

在测试的时候要注意设置的阈值，这对于实验的结果十分重要，阈值的设置很大程度上影响了数据的输出结果