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

Проверьте это:

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>

Answer 2

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"))

Я даже не проверял, скомпилируется ли он, но дайте или возьмите несколько опечаток, это должно сработать.