Spark code with explicit schemas - dangerously parametrised

parametricStatic.scala

def matchViewershipWithSchedules[T, RESULT](
  viewership: Dataset[T], schedules: Dataset[Schedule])(
  // You need this so that RESULT can be used as a type parameter
  // inside the body of the function. A quirk of how Scala conforms
  // to Java's type erasure rules
  implicit ev: TypeTag[RESULT]): Dataset[RESULT] = {
  import session.implicits._

  viewershipWithQHAndDate.toDF.as('viewership).join(schedules.toDF.as('schedules),
    $"viewership.date" === $"schedules.date"
      && $"viewership.segment_time" === $"schedules.segment_time"
      && $"viewership.station_id" === $"schedules.station_id")
    // filter on start and end time to avoid spurious matches
    .filter(
      $"schedules.start_time" < $"viewership.end_time")
        && $"schedules.end_time" > $"viewership.start_time")
    )
    .selectExpr("viewership.*", "schedules.program_id").as[RESULT]
}