How can we pass a variable to where clause in Spark Dataframe

user_13812346 0
Viewed 2592

I am trying to pass a variable SCD_filter to the where clause in the dataframe in spark for which I am receiving an error but when passed directly its working fine.I am doing this to dynamically pass this filter based on different scenarios for future use.

  val SCD_filter = """currentDF.col("u_business_unit") <=> updatedDF.col("u_business_unit")
                     |      and(currentDF.col("u_operation_level_2") <=> updatedDF.col("u_operation_level_2"))
                     |      and(currentDF.col("u_operation_level_3") <=> updatedDF.col("u_operation_level_3"))""".stripMargin

I am then passing the variable to the code below:

val common_unchangedata = currentDF.alias("currentDF")
.join(updatedDF, currentDF.col("Sys_id") === updatedDF.col("Sys_id"), "inner")
.select("currentDF.*")
.where(s"$SCD_filter")  /// passing the variable which is causing the error
.show()

Error Received:

Exception in thread "main" org.apache.spark.sql.catalyst.analysis.NoSuchDatabaseException: Database 'currentdf' not found;

Note: the currentDF is fine as the code is executing when variable is removed and we pass the condition to the where clause instaed of variable.enter image description here

apache-spark-sql apache-spark scala
3 Answers

try this -

val common_unchangedata = currentDF.alias("currentDF")
.join(updatedDF, currentDF.col("Sys_id") === updatedDF.col("Sys_id"), "inner")
.select("*")
.where($"Column1='$data'")  /// passing the variable which is causing the error
.show()
user_114209840

Create alias for both DataFrames & use alias like alias.column_name inside string.

val SCD_filter = """
   (
     (currentDF.u_business_unit <=> updatedDF.u_business_unit) and 
     (currentDF.u_operation_level_2 <=> updatedDF.colu_operation_level_2) and 
     (currentDF.u_operation_level_3 <=> updatedDF.u_operation_level_3)
   )
"""

val common_unchangedata = currentDF.alias("currentDF")
.join(updatedDF, currentDF.col("Sys_id") === updatedDF.col("Sys_id"), "inner")
.select("currentDF.*")
.where(SCD_filter)
.show()
user_85934140

Based on your comment

actually this code will be running for 5 tables and all tables will be having different columns and all conditions will be different..I am open for a better approach aswell

I would solve this problem with something like this,


import org.apache.spark.sql.functions._

val cond1 = $"u_business_unit"     <=> $"updatedDF.u_business_unit"
val cond2 = $"u_operation_level_2" <=> $"updatedDF.u_operation_level_2"
val cond3 = $"u_operation_level_3" <=> $"updatedDF.u_operation_level_3"

val SCD_filter = cond1.and(cond2).and(cond3)

val common_unchangedata = currentDF
.join(updatedDF, currentDF.col("Sys_id") === updatedDF.col("Sys_id"), "inner")
.where(SCD_filter) 
.show()

You could extend this to have a function getCondition(tableName:String):Column which constructs the appropriate condition in runtime based on which type of data you are working with.

user_11780280
Related