Brtdku

Is there a better way to write a filter of multiple conditions which are similar in nature on a spark dataframe.

Assuming df is a spark dataframe having timestamp columns 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"))) 

Any better way to write the same. Since I'm new to scala, I kind of don't know the best practices to code in scala yet. Appreciate any help.

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

I didn't even check if it compiles, but give or take a few typos it should do the job.

Check this out:

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 test case

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>