A somewhat working attempt at some generic spark code

genericSparkBroken.scala

trait TimeColumns {
  val date: Date
  val segment_time: Timestamp
  val station_id: Integer
  val start_time: Timestamp
  val end_time: Timestamp
}

case class Viewership(
  date: Date, segment_time: Timestamp, station_id: Integer, start_time: Timestamp, end_time: Timestamp, 
  viewership_property: String) with TimeColumns

case class Schedule(
  program_id: Integer, 
  date: Date, segment_time: Timestamp, station_id: Integer, start_time: Timestamp, end_time: Timestamp) with TimeColumns

case class ViewershipWithSchedule(
  date: Date, segment_time: Timestamp, station_id: Integer, start_time: Timestamp, end_time: Timestamp,
  viewership_property: String, program_id: Integer) with TimeColumns

def matchViewershipWithSchedules[T <: TimeColumns, 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")
    )
    // DANGER! if you pass in a ViewershipWithSchedule as the viewership, this will break
    .selectExpr("viewership.*", "schedules.program_id").as[RESULT]
}