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

【问题标题】：Get next week date in Spark Dataframe using scala使用 scala 在 Spark Dataframe 中获取下周日期
【发布时间】：2018-08-31 01:11:29
【问题描述】：

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

示例：输入：2017 年 1 月 1 日星期一输出：2017 年 1 月 2 日（星期二）输入：2017 年 3 月 4 日星期六输出：2017 年 3 月 5 日（星期一）

我已经通过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)
 }
}

需要帮助才能使其有效和工作。

【问题讨论】：

你的逻辑很好，但是你需要使用when()和otherwise()而不是if和else。这是编写 Spark 代码时的常见错误。请参阅我的答案以了解您的逻辑的功能版本。

标签： scala spark-dataframe

【解决方案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()
  }
}

【讨论】：

我是 spark-dataframe 的新手，我们可以在 spark-dataframe 中使用 java 库并像您所做的那样使用 scala 调用它吗？我怎么称呼这个 udf ，它似乎没有名字？
确实，不支持 LocalDate。您应该使用 java.sql.Date。无论如何，您可以将日期用作常规字符串并在您的 udf 中转换为 LocalDate。如我刚刚添加的示例所示，udf 可用于操作数据框的列。
我没听懂你。您说不支持 LocalDate，但我们在 udf 中使用 LocalDate。那么这是否意味着我们可以在 spark 数据帧中使用 java 库并使用 scala 调用它们？
如果你在你的 udf 中返回一个 LocalDate 类型，你会收到类似这样的消息：“不支持 java.time.LocalDate 类型的模式”。相反，如果您将 .toString 应用于您的 newDate 它将起作用。

【解决方案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 更详细地解释了这个函数的推导。

【讨论】：