UDF2

时间：2021-06-06 14:12:07

问题

  • 根据给定的gps点point(x,y)和北京的shape数据，关联出 AOI ID

IO

  • 输入
    • gps点表
    • create table gps
(
   x double, //经度
   y double //纬度
)

      +------------+------------+
      | x | y |
      +------------+------------+
      | 113.570517 | 22.149751 |
      | 113.57431 | 22.152167 |
      | 113.544971 | 22.190477 |
      | 113.546035 | 22.203194 |
      | 113.568557 | 22.168465 |

    • shape文件
    • ogrinfo -ro -al beijing_20160122152124568.shp

OGRFeature(beijing_20160122152124568):
  AOIID (String) = G00AOI000001ZIQ3
  POIID (String) = B0FFFTLAU4
  Name (String) = ˮѧԺԺ
  Type (String) =
  Disp_Class (Integer) =
  Reference (String) = (null)
  ParentID (String) = (null)
  CenterX (Real) = 116.322184
  CenterY (Real) = 39.931526
  Area (Real) = 19187.288500
  CreateTime (String) = -- :
  UpdateTime (String) = -- :
  Update (String) = u
  CityName (String) =   Memo (String) =   POLYGON ((418756.3848 143755.6608,418756.5396 143755.7688,418759.47 143755.7688,418760.8884 143755.7616,418760.8884 143754.616799999989,418761.9324 143754.5628,418763.556 143754.5376,418763.624399999972 143753.8716,418763.9304 143753.810399999988,418763.898 143751.5064,418756.395599999989 143751.733199999988,418756.395599999989 143753.263199999987,418756.3848 143755.6608))
  • 输出　　
    • 根据gps表中点，判断是否内含与shape文件的polygon中，关联出AOIID
    • 输出表
    • +------------+------------+--------+
| x          | y          | aoi_id |
+------------+------------+--------+
| 419217.2568 | 143807.9436 | NULL   |
| 418659.14879999997 | 143630.046 | NULL   |
| 418659.6312 | 143629.9848 | NULL   |
| 418024.2636 | 143054.75519999999 | NULL   |
| 419875.88399999996 | 143947.60199999998 | NULL   |
| 418042.9692 | 143037.2556 | NULL   |
| 418469.9436 | 144192.348 | NULL   |
| 418469.9436 | 144192.348 | G00AOI000001WBC2 |
| 420292.3248 | 143810.84879999998 | NULL   |
| 418509.26279999997 | 143762.994 | NULL   |

知识点

  • GDAL/ogr2ogr使用
  • JTS geometry内含判断
  • JTS Rtree使用

数据准备

  • shape文件
    • 提取wkt和aoiid字段
    • ogr2ogr -lco "GEOMETRY=AS_WKT" -lco "SEPARATOR=SEMICOLON" -f CSV beijing_20160122152124568.csv -sql "select AOIID from beijing_20160122152124568" beijing_20160122152124568.shp
    • 去掉文件中的双引号
    • sed -e '/"/s/"//g' beijing_20160122152124568.csv > beijing.csv
    • odps中创建表
    • create table beijing(wkt string,lua_id bigint);
    • 上传文件
    • tunnel u -fd ';' beijing.csv.csv autonavi_tinfo_dev.exercise_land_user_area; //列之间按照";"分割
    • 将表作为资源文件加入odps
    • add table beijing

编码

  • udtf
  • import java.util.Iterator;
import java.util.List;

import com.aliyun.odps.udf.ExecutionContext;
import com.aliyun.odps.udf.UDFException;
import com.aliyun.odps.udf.UDTF;
import com.aliyun.odps.udf.annotation.Resolve;
import com.vividsolutions.jts.geom.Coordinate;
import com.vividsolutions.jts.geom.Envelope;
import com.vividsolutions.jts.geom.GeometryFactory;
import com.vividsolutions.jts.geom.Point;
import com.vividsolutions.jts.geom.Polygon;
import com.vividsolutions.jts.index.strtree.STRtree;
import com.vividsolutions.jts.io.WKTReader;

/**
 * 加载table中的资源，字段如： wkt string ,bigint lua_id 参数： x,y,tableName,idxWkt,idxId
 *
 * @author xingxing.dxx double,double,string,bigint,bigint->double,double,string
 *
 */
@Resolve({ "double,double->double,double,string" })
public class ExerciseWithinUDTF extends UDTF {

    /*
     * private Logger LOGGER =
     * LoggerFactory.getLogger(ExerciseWithinUDTF.class);
     */

    private GeometryFactory geometryFactory = new GeometryFactory();
    private WKTReader reader = new WKTReader(geometryFactory);
    private STRtree strtree;
    private ExecutionContext ctx;

    public void setup(ExecutionContext ctx) throws UDFException {
        this.ctx = ctx;
    }

    @Override
    public void process(Object[] args) throws UDFException {

        final double x = (double) args[0];
        final double y = (double) args[1];
        String tableName = (String) args[2];
        int idxWkt = ((Long) args[3]).intValue();
        int idxId = ((Long) args[4]).intValue();

        // 1.初始化，第一次才加载表资源
        if (strtree == null) {
            strtree = new STRtree();
            try {
                Iterator<Object[]> iterator = this.ctx.readResourceTable(tableName).iterator();
                while (iterator.hasNext()) {
                    Object[] o = iterator.next();
                    String wkt = (String) o[idxWkt];
                    String id = (String) o[idxId];
                    Polygon polygon = (Polygon) reader.read(wkt);
                    GeometryBean gb = new GeometryBean(polygon, id);
                    strtree.insert(polygon.getEnvelopeInternal(), gb);
                }

            } catch (Exception e) {
                /* LOGGER.error("资源加载出现问题",e); */
                e.printStackTrace();
                throw new RuntimeException(e.getMessage());

            }
        }

        Coordinate coord = new Coordinate(x, y);
        Point point = geometryFactory.createPoint(coord);
        Envelope envelope = new Envelope(coord);
        List<GeometryBean> items = strtree.query(envelope);
        String luaId = null;
        if (items != null && items.size() > 0) {
            for (Object o : items) {
                GeometryBean geometryBean = (GeometryBean) o;
                if (point.within(geometryBean.getPolygon())) {
                    luaId = geometryBean.getId();
                }
                forward(x, y, luaId);
            }
        } else {
            forward(x, y, luaId);
        }

    }

    public void close() throws UDFException {

    }

}
  • sql
  • CREATE TABLE result_gps_within
AS
SELECT
    udtf_exercise_within(x,y,,)
    as(x,y,aoi_id)
   FROM gps 
  • SELECT a.aoi_id,count(distinct x,y) num
  FROM
     (
      SELECT
          udtf_exercise_within(x*,y*,,)
          as(x,y,aoi_id)
        FROM gps
     ) a where aoi_id is not null
 GROUP BY a.aoi_id

碰到的坑

  • shape文件是秒的格式，需要转换成度
  • ogr2ogr转成的csv文件需要去掉双引号，否则不能反序列化为polygon
  • udtf读取资源文件需要把表当做资源添加

相关文章