Spark SQL 筛选多个相似的字段

Question

有没有更好的方法在火花数据帧上编写性质相似的多个条件的过滤器。

假设 df 是具有时间戳列 t1，t2，t3，t4 的火花数据帧。

val filteredDF=df.filter(col("t1").lt(current_date()-expr("INTERVAL 30 DAYS")) || col("t2").lt(current_date()-expr("INTERVAL 30 DAYS")) ||
col("t3").lt(current_date()-expr("INTERVAL 30 DAYS")) ||
col("t4").lt(current_date()-expr("INTERVAL 30 DAYS"))) 

任何更好的方法来写同样的东西。由于我是 scala 的新手，我还不知道用 scala 编码的最佳实践。感谢任何帮助。

Answer 1

import df.sparkSession.implicits._
import org.apache.spark.sql.functions._
def filterDates(dates: Column*): Column = 
  dates
    .map(_.lt(current_date()-expr("INTERVAL 30 DAYS")))
    .reduce(_ or _)
val filteredDF = df.filter(filterDates($"t1", $"t2", $"t3", $"t4"))

我什至没有检查它是否编译，但给出或接受一些错别字它应该可以完成这项工作。

Answer 2

看看这个：

scala>  val df =Seq( ( (Timestamp.valueOf("2019-01-01 01:02:03")), (Timestamp.valueOf("2019-01-10 01:02:03")), (Timestamp.valueOf("2019-01-15 01:02:03") ), (Timestamp.valueOf("2019-02-22 01:02:03")) ) ).toDF("t1","t2","t3","t4")
df: org.apache.spark.sql.DataFrame = [t1: timestamp, t2: timestamp ... 2 more fields]
scala> df.show(false)
+-------------------+-------------------+-------------------+-------------------+
|t1                 |t2                 |t3                 |t4                 |
+-------------------+-------------------+-------------------+-------------------+
|2019-01-01 01:02:03|2019-01-10 01:02:03|2019-01-15 01:02:03|2019-02-22 01:02:03|
+-------------------+-------------------+-------------------+-------------------+

scala> val ts_cols = df.dtypes.filter( _._2 == "TimestampType" ).map( _._1)
ts_cols: Array[String] = Array(t1, t2, t3, t4)
scala> val exp1 = ts_cols.map ( x=> col(x).lt(current_date()-expr("INTERVAL 30 DAYS")) ).reduce( _||_ )
exp1: org.apache.spark.sql.Column = ((((t1 < (current_date() - interval 4 weeks 2 days)) OR (t2 < (current_date() - interval 4 weeks 2 days))) OR (t3 < (current_date() - interval 4 weeks 2 days))) OR (t4 < (current_date() - interval 4 weeks 2 days)))
scala> df.select(col("*"),exp1.as("ts_comp") ).show(false)
+-------------------+-------------------+-------------------+-------------------+-------+
|t1                 |t2                 |t3                 |t4                 |ts_comp|
+-------------------+-------------------+-------------------+-------------------+-------+
|2019-01-01 01:02:03|2019-01-10 01:02:03|2019-01-15 01:02:03|2019-02-22 01:02:03|false  |
+-------------------+-------------------+-------------------+-------------------+-------+

true测试用例

scala> val df2 =Seq( ( (Timestamp.valueOf("2018-01-01 01:02:03")), (Timestamp.valueOf("2018-01-10 01:02:03")), (Timestamp.valueOf("2018-01-15 01:
02:03") ), (Timestamp.valueOf("2018-02-22 01:02:03")) ) ).toDF("t1","t2","t3","t4")
df2: org.apache.spark.sql.DataFrame = [t1: timestamp, t2: timestamp ... 2 more fields]
scala> df2.select(col("*"),exp1.as("ts_comp") ).show(false)
+-------------------+-------------------+-------------------+-------------------+-------+
|t1                 |t2                 |t3                 |t4                 |ts_comp|
+-------------------+-------------------+-------------------+-------------------+-------+
|2018-01-01 01:02:03|2018-01-10 01:02:03|2018-01-15 01:02:03|2018-02-22 01:02:03|true   |
+-------------------+-------------------+-------------------+-------------------+-------+

scala>