SpatialKeyComparition.scala

SpatialKeyComparition.scala

object IngestImage extends App{


val _spark = SparkSession
    .builder()
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.kryo.registrator", "geotrellis.spark.io.kryo.KryoRegistrator")
    .config("spark.logConf", "true")
    .master("local[*]")
    .appName("SpatialKeyComparition")
    .config("spark.ui.enabled",true)
    .config("spark.executor.memory","8g")
    .config("spark.driver.memory","9g")
    .getOrCreate()

  import _spark.implicits._
  
  try {

      val conf = _spark.sparkContext.getConf
      implicit val sc = _spark.sparkContext
      implicit val _sql = _spark.sqlContext
      
       val extent = Extent(-109, 37, -102, 41) // Extent of Colorado

      def randomPointFeature(extent: Extent) = {
        def randInRange (low: Double, high: Double): Double = {
          val x = Random.nextDouble
          low * (1-x) + high * x
        }
        MultiLine(Array(Line(Array(Point(randInRange(extent.xmin, extent.xmax),      // the geometry
          randInRange(extent.ymin, extent.ymax)), Point(randInRange(extent.xmin, extent.xmax),      // the geometry
          randInRange(extent.ymin, extent.ymax))))))
      }

      val linesArray = (for (i <- 1 to 100000) yield randomPointFeature(extent)).toList

      val lines = sc.parallelize ( linesArray.map(_.reproject(LatLng, WebMercator)), 200 )
      
       val tileSize = 512
      val layout = ZoomedLayoutScheme(WebMercator, tileSize).levelForZoom(9).layout




      def lineToSpatialKey(l: Geometry): Iterator[(SpatialKey, (Geometry, SpatialKey))] = {
        val lineextent = l.envelope
        //var keySet = Set.empty[SpatialKey]
        val gridBounds = layout.mapTransform(lineextent)
        val r =  for {
          (c, r) <- gridBounds.coords
          if r < layout.tileLayout.totalRows
          if c < layout.tileLayout.totalCols
        } yield ( (SpatialKey(c, r), (l, SpatialKey(c, r))))
        r.toIterator
      }


      /** Key geometry by spatial keys of intersecting tiles */
      def keyGeom(geom: Geometry): Iterator[(SpatialKey, (Geometry, SpatialKey))] = {
        val lineextent = geom.envelope
        //var keySet = Set.empty[SpatialKey]
        val gridBounds = layout.mapTransform(lineextent)

        val layoutRasterExtent = RasterExtent(layout.extent, layout.layoutCols, layout.layoutRows)
        val layoutRasterizerOptions = Rasterizer.Options(includePartial=true, sampleType=PixelIsPoint)
        var keySet = Set.empty[SpatialKey]
        geom.foreach(layoutRasterExtent, layoutRasterizerOptions){ (col, row) =>
          keySet = keySet + SpatialKey(col, row)
        }
        keySet.toIterator.map { key => (key, (geom, key)) }
      }

//compare the two spatial keys collecction
      val ree = lines.map{x=>
        val key = keyGeom(x)
        val lineS= lineToSpatialKey(x)
        (key.toList, lineS.toList)
      }.collect()
      
      
      
      
     
       } finally {
      _spark.stop()
    }
    }



}