swkimme

  def distanceByQuadtree(sc: SparkContext, inputDF: DataFrame, lookupRDD: RDD[(Geometry, String)], count: Long, distance: Double, bulkSize: Int) : RDD[(Geometry, String)] = {
    QuadtreeHandler.rddInit()
    QuadtreeHandler.setBulkSize(bulkSize)
    QuadtreeHandler.setRDDCount(count)
    val rddWithIndex = lookupRDD.zipWithIndex().map(_.swap).cache()
    var totalCount = 0L
    var iter = 0
    var result:RDD[(Geometry, String)] = null
    var (qdTree, env) = QuadtreeHandler.getQuadtreeAndEnvelope(rddWithIndex, distance)
    while( qdTree != null ) {