Spark Schema Comparison

comparision.scala

import org.apache.spark.sql.types._
import cats._
import cats.implicits._

sealed trait ComparationResult 
case object Validated extends ComparationResult
case class Invalid(msg: String) extends ComparationResult

def safeGetColumn(dfSchema: StructType,columnName: String) = 
try{
  Some(dfSchema(columnName))
}catch {
 case x: IllegalArgumentException => None
 case x : Throwable => throw x
}

def matchDFSchemas(df1: DataFrame,df2: DataFrame) = {
  val df1Columns = df1.schema.map(_.name).toSet
  val df2Columns = df2.schema.map(_.name).toSet
  (df1Columns union df2Columns).map{columnName =>
   val df1Col = safeGetColumn(df1.schema,columnName)
   val df2Col = safeGetColumn(df2.schema,columnName)
   (df1Col,df2Col).mapN{(col1,col2) =>
   if (col1.dataType != col2.dataType)
     Invalid(msg = String.format("Different types on Column: %s. df1 type: %s and df2 type %s:",columnName,col1.dataType,col2.dataType))
   else Validated
   }.getOrElse(Invalid(msg = String.format("Column doesn't exists on both DataFrames: %s",columnName)))
  }
}