使用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()
  }
}

Answer 2

spark-daria 中定义了一个更简单的答案：

def nextWeekday(col: Column): Column = {
  val d = dayofweek(col)
  val friday = lit(6)
  val saturday = lit(7)
  when(col.isNull, null)
    .when(d === friday || d === saturday, next_day(col,"Mon"))
    .otherwise(date_add(col, 1))
}

您总是希望尽可能坚持使用本机 Spark 函数。 post 更详细地解释了此函数的推导。

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

Get next week date in Spark Dataframe using scala

scala

spark-dataframe