GeodistanceCalculationTest.scala

package com.ta.geo.calculation

import com.holdenkarau.spark.testing.DataFrameSuiteBase
import com.ta.geo.calculation.GeodistanceCalculation.matchTrackingEventsWithNearestAirport
import com.ta.geo.datasource.ModelDatasource
import com.ta.geo.measure.{DistanceMeasure, HaversineDistance}
import com.ta.geo.model.{Airport, TrackingEvent}
import org.apache.spark.sql.{Dataset, SparkSession}
import org.scalatest.{FlatSpec, Matchers}

class GeodistanceCalculationTest extends FlatSpec with Matchers with DataFrameSuiteBase {

  override implicit def reuseContextIfPossible: Boolean = true

  implicit val sqlCtx: SparkSession = spark
  import sqlCtx.implicits._

  val ds: ModelDatasource = new ModelDatasource // uses sparksession to read dataset
  val airports: Dataset[Airport] = ds.getAirports("data/optd-sample-20161201.csv")

  "GeolocationCalculation" should "return Berlin Tempelhof Airport (THF)." in {
    val events = Seq(TrackingEvent("TravelAudience", 52.4897337, 13.4554343)).toDS()

    val actual = matchTrackingEventsWithNearestAirport(events, airports, new HaversineDistance)
    val expected = Seq(("TravelAudience", "THF")).toDF("uuid", "iata_code")

    assertDataFrameEquals(actual, expected)
  }
  
}

and everything works fine if I do it in a manual way (but then I don't have your DS/DF assertion methods and all other goodies) :(

class GeodistanceCalculationTest extends FlatSpec with Matchers with BeforeAndAfter {

  implicit val spark: SparkSession =
    SparkSession.builder.master("local[*]")
      .config("spark.sql.shuffle.partitions", "4")
      .config("spark.driver.allowMultipleContexts", "true")
      .getOrCreate

  import spark.implicits._

  before {
    spark.newSession()
  }

  after {
    System.clearProperty("spark.driver.port")
  }

  val ds = new ModelDatasource // uses spark
  val airports = ds.getAirports("data/optd-sample-20161201.csv")

  "GeolocationCalculation" should "return Zagreb Airport (ZAG)." in {
    val events = Seq(TrackingEvent("Filip", 45.811289, 16.051475)).toDS

    val actual = matchTrackingEventsWithNearestAirport(events, airports, new HaversineDistance)
    val expected = Seq(("Filip", "ZAG")).toDF("uuid", "iata_code")

    actual.collect() shouldEqual expected.collect()
  }
}