如何在火花中检查列名和与之关联的数据是否匹配scala

Question

假设我有以下几列：

EMP_ID, EMP_NAME, EMP_CONTACT  
1, SIDDHESH, 544949461

现在我想验证数据是否与列名模式同步。对于EMP_NAME，该列中的数据应该仅为string。我在引用此链接后尝试了下面的代码，但它在我的代码的最后一行显示错误。

package com.sample
import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.Row
class sample1 {
  val spark = SparkSession.builder().master("local[*]").getOrCreate()
  val data = spark.read.format("csv").option("header", "true").load("C:/Users/siddheshk2/Desktop/words.txt")
  val originalSchema = data.schema

  def validateColumns(row: Row): Row = {
    val emp_id = row.getAs[String]("EMP_ID")
    val emp_name = row.getAs[String]("EMP_NAME")
    val emp_contact = row.getAs[String]("EMP_CONTACT")

    // do checking here and populate (err_col,err_val,err_desc) with values if applicable

    Row.merge(row)
  }
  val validateDF = data.map { row => validateColumns(row) }

}  

因此，它不接受代码的最后一行。映射{行=

我输入了一条无效记录（第三条），如下所示：

EMP_ID,EMP_NAME,EMP_CONTACT
1,SIDDHESH,99009809
2,asdadsa, null
sidh,sidjh,1232  

在本例中，我已经为id列输入了一个字符串值（应该是一个数字），因此在检查列架构及其数据后，它应该抛出一个错误，说明记录与列架构不匹配。

Answer 1

行类有一个模式属性。您可以通过遍历列并比较它们来使用它。为此，您可以使用==运算符或使用此处描述的模式比较方法之一。

然后验证方法可以看起来像下一个：

def isValid(row: Row): Boolean = originalSchema == row.schema

Answer 2

您刚刚错过了将数据帧转换为rdd以应用<代码>。映射操作请尝试以下操作：

import org.apache.spark.sql.Row    
val validateDF = data.rdd.map { row => validateColumns(row) }

如果要将其转换回DataFrame，只需使用sparkSession即可：

val newSchema = // specify the schema of the new dataframe
val updatedDF = spark.createDataFrame(validateDF, newSchema)