val spark = SparkSession.builder
.appName("Testing Spark DSL")
.master("local[1]") //build a local cluster
.getOrCreate()
// injectRules(spark)
import spark.implicits._
val data = Array(("US", "TX", "2018-12-08 00:00:00", 12.0123, "ios", 2, 32.813548, -96.835159),
("US", "PA", "2018-12-08 00:00:00", 12.0123, "ios", 183,32.813548, -96.835159),
("CA", null, "2018-12-08 00:00:00", 12.0123, "android", 183,32.813548, -96.835159),
("GB", null, "2018-12-08 00:00:00", 12.0123, "ios", 2,32.813548, -96.835159),
("US", "NC", "2018-12-08 00:00:00", 12.0123, "android", 35,32.813548, -96.835159),
("US", "CA", "2018-12-08 00:00:00", 12.0123, null, 2,32.813548, -96.835159),
("A", null, "2018-12-08 00:00:00", 12.0123, "android", 183,32.813548, -96.835159),
("US", "NY", "2018-12-08 00:00:00", 12.0123, "ios", 2, 32.813548, -96.835159))
val df1 = spark.sparkContext.parallelize(data).toDF("country", "state", "location_at",
"horizontal_accuracy", "platform", "app_id", "latitude", "longitude")
.withColumn("location_at", col("location_at").cast(TimestampType))
df1.show()
println(df1.printSchema)
val filterFilePath = path_to_geojson
val filteringDS = spark.sqlContext.read.format("magellan")
.option("magellan.index", "true")
.option("magellan.index.precision", "15")
.option("type", "geojson").load(filterFilePath)
.cache()
filteringDS.count()
filteringDS.show(false)
val filtered = df1
.withColumn("locationPoint", point(col("longitude"), col("latitude")))
.join(filteringDS)
.where(col("locationPoint") within col("polygon"))
filtered.show()
Also, to note, I've tried different levels of precision in the index but the same issue persisted when injecting the rules.
Using the example above, if I just
injectRulesI get 0 results. But if I don't useinjectRulesI get the proper results.Also, to note, I've tried different levels of precision in the index but the same issue persisted when injecting the rules.
Geojson file used for testing attached.
TX.geojson.txt