scala - 使用scala在Spark Dataframe中获取下周日期

Question

我在函数中有一个 DateType 输入。我想排除星期六和星期日并得到下一个工作日，如果输入日期是周末，否则它应该给出第二天的日期

例子:输入:星期一 1/1/2017 输出:1/2/2017(即星期二)输入:星期六 3/4/2017 输出:3/5/2017(即星期一)

我已经经历了https://spark.apache.org/docs/2.0.2/api/java/org/apache/spark/sql/functions.html但我没有看到现成的功能，所以我认为需要创建它。

到目前为止，我有一些东西是:

val nextWeekDate = udf {(startDate: DateType) =>
val day= date_format(startDate,'E'
 if(day=='Sat' or day=='Sun'){
  nextWeekDate = next_day(startDate,'Mon')
}
 else{
   nextWeekDate =   date_add(startDate, 1)
 }
}

需要帮助才能使其有效并正常工作。

Answer 1

使用日期作为字符串:

import java.time.{DayOfWeek, LocalDate}
import java.time.format.DateTimeFormatter

// If that is your format date
object MyFormat {
  val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd")
}

object MainSample {
  import MyFormat._

  def main(args: Array[String]): Unit = {
    import java.sql.Date
    import org.apache.spark.sql.types.{DateType, IntegerType}

    import spark.implicits._  

    import org.apache.spark.sql.types.{ StringType, StructField, StructType }
    import org.apache.spark.sql.functions._

    implicit val spark: SparkSession =
          SparkSession
            .builder()
            .appName("YourApp")
            .config("spark.master", "local")
            .getOrCreate()

    val someData = Seq(
      Row(1,"2013-01-30"),
      Row(2,"2012-01-01")
    )

    val schema = List(StructField("id", IntegerType), StructField("date",StringType))
    val sourceDF = spark.createDataFrame(spark.sparkContext.parallelize(someData), StructType(schema))

    sourceDF.show()

    val _udf = udf { (dt: String) =>
      // Parse your date, dt is a string
      val localDate = LocalDate.parse(dt, formatter)

      // Check the week day and add days in each case
      val newDate = if ((localDate.getDayOfWeek == DayOfWeek.SATURDAY)) {
        localDate.plusDays(2)
      } else if (localDate.getDayOfWeek == DayOfWeek.SUNDAY) {
        localDate.plusDays(1)
      } else {
        localDate.plusDays(1)
      }
      newDate.toString
    }

    sourceDF.withColumn("NewDate", _udf('date)).show()
  }
}