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 WKBReade
        //copy the byte array, removing the first fou
        //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