SparkSQL使用UDF函数代替MySQL空间函数读取MySQL空间字段

  @throws[Exception]
  def sparkUDFSTAsText(geometryAsBytes: Array[Byte]): Geometry = {
    var dbGeometry: Geometry = null
    if (geometryAsBytes.length < 5) throw new Exception("Invalid geometry inputStream - less than five bytes")
    //first four bytes of the geometry are the SRID,
    //followed by the actual WKB.  Determine the SRID
    //here
    val sridBytes = new Array[Byte](4)
    System.arraycopy(geometryAsBytes, 0, sridBytes, 0, 4)
    val bigEndian: Boolean = geometryAsBytes(4) == 0x00
    var srid = 0
    if (bigEndian) for (i <- 0 until sridBytes.length) {
      srid = (srid << 8) + (sridBytes(i) & 0xff)
    }
    else for (i <- 0 until sridBytes.length) {
      srid += (sridBytes(i) & 0xff) << (8 * i)
    }
    //use the JTS WKBReader for WKB parsing
    val wkbReader = new WKBReader
    //copy the byte array, removing the first four
    //SRID bytes
    val wkb = new Array[Byte](geometryAsBytes.length - 4)
    System.arraycopy(geometryAsBytes, 4, wkb, 0, wkb.length)
    dbGeometry = wkbReader.read(wkb)
    dbGeometry.setSRID(srid)
    dbGeometry
  }

    public Geometry sparkUDFSTAsText(byte[] geometryAsBytes) throws Exception {
        Geometry dbGeometry = null;
        if (geometryAsBytes.length < 5) throw new Exception("Invalid geometry inputStream - less than five bytes");
        byte[] sridBytes = new byte[4];
        System.arraycopy(geometryAsBytes, 0, sridBytes, 0, 4);
        boolean bigEndian = geometryAsBytes[4] == 0x00;
        int srid = 0;
        if(bigEndian) {
            for(int i=0; i<sridBytes.length; i++) {
                srid = (srid << 8) + (sridBytes[i] & 0xff);
            }
        } else {
            for(int i=0; i<sridBytes.length; i++) {
                srid += (sridBytes[i] & 0xff) << (8 * i);
            }
        }
        WKBReader wkbReader = new WKBReader();
        byte[] wkb = new byte[geometryAsBytes.length - 4];
        System.arraycopy(geometryAsBytes, 4, wkb, 0, wkb.length);
        dbGeometry = wkbReader.read(wkb);
        dbGeometry.setSRID(srid);
        return dbGeometry;
    }

    def toGeometryText(binary: Array[Byte]) = sparkUDFSTAsText(binary).toText
    spark.udf.register("ST_ASTEXT",toGeometryText(_))
    val rddROW: RDD[Row] = spark.sql("SELECT id, ST_ASTEXT(point), ST_ASTEXT(polygon) FROM t_point_polygon").limit(10).rdd