Q:

Apply validation with some columns in Spark SQL

Validation dataframe:
+---------+---------------------------+-------------------------+
|dataframe|Validation Checks          |cols                     |
+---------+---------------------------+-------------------------+
|Attendee |isEmpty,IsNull             |col1,col2,col3           |
+---------+---------------------------+-------------------------+

Attendee dataframe:
    col1    col2    col3
    a1       a2     a3
             b2     b3
    c1       c2     c3
    d1       d2     d3

Expected result dataframe:
    col1    col2    col3   status
    a1       a2     a3      clean
             b2     b3      dirty  
    c1       c2     c3      clean
    d1       d2     d3      clean

Code used:
var columns = df.columns //struct(df.columns map col: _*) 
val colDF = df.select(col("dataframe")) 
var tablename = colDF.head().toSeq 
val checkDF = df.select(col("Validation Checks")) 
val opsColDF = df.select(col("cols")) 
val opsColumn = opsColDF.columns println("opsColumn :::" + opsColumn) 

A:

If you have a dataframe as 
+---------+-----------------+--------------+
|dataframe|Validation Checks|cols          |
+---------+-----------------+--------------+
|Attendee |isEmpty,isNull   |col1,col2,col3|
+---------+-----------------+--------------+

You should make sql query using the column values. I have created another column using udf function making valid query 
import org.apache.spark.sql.functions._
def createQueryUdf = udf((table: String, logic: String, cols: String) => {
  "select *, case when "+
    cols.split(",")
      .map(_.trim)
      .map(x => logic.split(",")
        .map(_.trim.toLowerCase)
        .map{
            case y if (y == "isempty") => s"$x like ''"
            case y => s"$y($x)"
        }.mkString(" or "))
      .mkString(" or ") +
  s" then 'dirty' else 'clean' end as status from $table"
})

val dataframeWithQuery = df.withColumn("query", createQueryUdf(col("dataframe"), col("Validation Checks"), col("cols")))

so dataframeWithQuery would be 
+---------+-----------------+--------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|dataframe|Validation Checks|cols          |query                                                                                                                                                                 |
+---------+-----------------+--------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|Attendee |isEmpty,isNull   |col1,col2,col3|select *, case when col1 like '' or isnull(col1) or col2 like '' or isnull(col2) or col3 like '' or isnull(col3) then 'dirty' else 'clean' end as status from Attendee|
+---------+-----------------+--------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------+

Now you can select the valid query to hit on the dataframes but before that the dataframes are all registered as 
attendee.createOrReplaceTempView("Attendee")

Then you can just collect the query column and loop to apply the query statements
val queryArray = dataframeWithQuery.select("query").collect.map(_.getAs[String]("query"))

for(query <- queryArray){
  spark.sql(query).show(false)
}

which should give you 
+----+----+----+------+
|col1|col2|col3|status|
+----+----+----+------+
|a1  |a2  |a3  |clean |
|    |b2  |b3  |dirty |
|c1  |c2  |c3  |clean |
|d1  |d2  |d3  |clean |
+----+----+----+------+

By now you should have idea on how to proceed further. I hope the answer is helpful

