转换字符串并与 DF 列值比较 Spark Scala

Question

我有一个val current_dates = "A:2021-04-02,B:2021-04-02,C:2021-04-01,D:2021-04-01" 类型的字符串。这里A,B,C,D是id字段及其对应的日期。

现在我有一个输入数据框，其中包含多个记录，其中包含 id 和日期列。

val input_df = sc.parallelize(Seq(("A","2021-04-01"),("A","2021-04-02"),("B","2021-04-01"),("B","2021-04-02"),("C","2021-04-01"),("C","2021-04-02"),("D","2021-04-01"),("D","2021-04-02"))).toDF("id","create_date")

input_df.show()

+---+-----------+
| id|create_date|
+---+-----------+
|  A| 2021-04-01|
|  A| 2021-04-02|
|  B| 2021-04-01|
|  B| 2021-04-02|
|  C| 2021-04-01|
|  C| 2021-04-02|
|  D| 2021-04-01|
|  D| 2021-04-02|
+---+-----------+

现在我想将每条记录的日期值与字符串中每个 id 的对应日期进行比较，并在 dataFrame 中导出新的日期列。

expected_df.select((input_df.columns ++ Array("new_dt")).head, (input_df.columns ++ Array("new_dt")).tail: _*).orderBy("id").show()
+---+-----------+----------+
| id|create_date|    new_dt|
+---+-----------+----------+
|  A| 2021-04-01|2021-04-02|
|  A| 2021-04-02|2021-04-02|
|  B| 2021-04-02|2021-04-02|
|  B| 2021-04-01|2021-04-02|
|  C| 2021-04-02|2021-04-02|
|  C| 2021-04-01|2021-04-01|
|  D| 2021-04-01|2021-04-01|
|  D| 2021-04-02|2021-04-02|
+---+-----------+----------+

目前，我正在将字符串转换为另一个数据帧，并将其与输入数据帧连接，并以下列方式派生新列。

val current_dates_df = sc.parallelize(current_dates.split(",").map(_.split(":")).map{ case Array(a,b) => (a, b) }).toDF("previous_run_id", "previous_run_date")    

current_dates_df.show()   

+---------------+-----------------+
|previous_run_id|previous_run_date|
+---------------+-----------------+
|              A|       2021-04-02|
|              B|       2021-04-02|
|              C|       2021-04-01|
|              D|       2021-04-01|
+---------------+-----------------+ 

val deriveNewDt: UserDefinedFunction = udf[String, String, String]((create_date: String, previous_run_date: String) => {
    val date_format: String = "yyyy-MM-dd"
    val new_dt = {
        if (new SimpleDateFormat(date_format).parse(create_date).after(new SimpleDateFormat(date_format).parse(previous_run_date))) create_date 
        else previous_run_date
    }
    new_dt
})    


val joined_df = input_df.join(current_dates_df, input_df("id") === current_dates_df("previous_run_id"), "left_outer")    

val expected_df = joined_df.withColumn("new_dt", deriveNewDt($"create_date", $"previous_run_date"))    

expected_df.select((input_df.columns ++ Array("new_dt")).head, (input_df.columns ++ Array("new_dt")).tail: _*).show()

是否有更好的方法来处理字符串并在不将字符串转换为 DataFrame 的情况下实现相同的功能。

Answer 1

您可以使用str_to_map 获取给定id 的日期，并使用greatest 获取两者之间的较晚日期：

val current_dates = "A:2021-04-02,B:2021-04-02,C:2021-04-01,D:2021-04-01"

val result = input_df.withColumn(
    "new_dt", 
    expr(s"greatest(str_to_map('$current_dates, ',', ':')[id], create_date)")
)

result.show
+---+-----------+----------+
| id|create_date|    new_dt|
+---+-----------+----------+
|  A| 2021-04-01|2021-04-02|
|  A| 2021-04-02|2021-04-02|
|  B| 2021-04-01|2021-04-02|
|  B| 2021-04-02|2021-04-02|
|  C| 2021-04-01|2021-04-01|
|  C| 2021-04-02|2021-04-02|
|  D| 2021-04-01|2021-04-01|
|  D| 2021-04-02|2021-04-02|
+---+-----------+----------+