python to scala

pyToScala.scala

def nearest_holiday(date):
    last_holiday = index.value[0]
    for next_holiday in index.value:
        if next_holiday >= date:
            break
        last_holiday = next_holiday
    if last_holiday > date:
        last_holiday = None
    if next_holiday < date:
        next_holiday = None
    return (last_holiday, next_holiday)


from pyspark.sql.types import *
return_type = StructType([StructField('last_holiday', StringType()), StructField('next_holiday', StringType())])

from pyspark.sql.functions import udf
nearest_holiday_udf = udf(nearest_holiday, return_type)
// ####################################### END original 
// now scala

val index = spark.sparkContext.broadcast(holidays)
case class HolidaysNearestDate(previous: Date, next: Date)

val holidays: (Date => HolidaysNearestDate) = (aktDate: Date) => {
    var last_holiday = Option(index.value(0).toLocalDate)
    val current = aktDate.toLocalDate
    //    var next_holiday = indexAT.value.last
    breakable {
      for (next_holiday <- indexAT.value) {
        val next = next_holiday.toLocalDate
        println("next ", next)
        println("last ", last_holiday)

        if (next.isAfter(current) || next.equals(current)) break
        // check do I actually get here?
        last_holiday = Option(next)
      } // TODO this is so not scala and ugly ...
      if (last_holiday.isDefined) {
        if (last_holiday.get.isAfter(current)) {
          last_holiday = None
        }
      }
      if (last_holiday.isDefined) {
        if (last_holiday.get.isBefore(current)) {
          // TODO use one more var because out of scope
          next = None
        }
      }
    }
    HolidaysNearestDate(Option(Date.valueOf(last_holiday.get)), Date.valueOf(next))
  }