据我所知，你不能这样做与内置的csv解析器。如果解析器遇到错误（failfast模式），您可以让它停止，但不能进行注解。
但是，您可以使用定制的csv解析器来实现这一点，它可以在一次传递中处理数据。除非我们想做一些聪明的类型自省，否则最简单的方法是创建一个helper类来注解文件的结构：

case class CSVColumnDef(colPos: Int, colName: String, colType: String)
val columns = List(CSVColumnDef(0,"fname","String"),CSVColumnDef(1,"lname","String"),CSVColumnDef(2,"age", "Int"))

接下来，我们需要一些函数来a）分割输入，b）从分割的数据中提取数据，c）检查行是否错误：

import scala.util.Try
def splitToSeq(delimiter: String) = udf[Seq[String],String](_.split(delimiter))
def extractColumnStr(i: Int) = udf[Option[String],Seq[String]](s => Try(Some(s(i))).getOrElse(None))
def extractColumnInt(i: Int) = udf[Option[Int],Seq[String]](s => Try(Some(s(i).toInt)).getOrElse(None))
def isRowBad(delimiter: String) = udf[Boolean,String](s => {
   (s.split(delimiter).length != columns.length) || (s.split(delimiter).exists(_.length==0))
})

要使用这些，我们首先需要读入文本文件（因为我没有文本文件，为了允许人们复制这个答案，我将创建一个rdd）：

val input = sc.parallelize(List(("will,smith,40"),("Dwayne,Nunn,36"),("Aniruddha,Sinha,"),("Maria,,22")))
input.take(5).foreach(println)

有了这个输入，我们就可以创建一个带有单列的Dataframe，即原始行，并将拆分列添加到其中：

val delimiter = ","
val raw = "raw"
val delimited = "delimited"
val compDF = input.toDF(raw).withColumn(delimited, splitToSeq(delimiter)(col(raw)))

最后，我们可以提取前面定义的所有列，并检查行是否错误：

val df = columns.foldLeft(compDF){case (acc,column) => column.colType match {
   case "Int" => acc.withColumn(column.colName, extractColumnInt(column.colPos)(col(delimited)))
   case _ => acc.withColumn(column.colName, extractColumnStr(column.colPos)(col(delimited)))
}}.
withColumn("isMyRowBad", isRowBad(delimiter)(col(raw))).
drop(raw).drop(delimited)
df.show
df.printSchema

这个解决方案的优点是spark execution planner足够聪明，可以构建所有这些 .withColumn 单通道操作( map )在数据上，没有零洗牌。恼人的是，与使用漂亮闪亮的csv库相比，开发人员的工作量要大得多，我们需要以某种方式定义列。如果您想更聪明一点，可以从文件的第一行获取列名（提示： .mapPartitionsWithIndex )，并将所有内容解析为字符串。我们也不能定义一个case类来描述整个df，因为有太多的列以这种方式处理解决方案。希望这有帮助。。。

这可以使用udf来完成。尽管ben horsburgh给出的答案肯定是非常出色的，但是我们可以做到这一点，而不必深入研究Dataframe背后的内部架构。
下面的代码可以给你一个想法

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
/**
  * Created by vaijnath on 10/4/17.
  */
object DataQualityCheck extends App {
  val spark = SparkSession.builder().master("local[*]").getOrCreate()
  import spark.implicits._
  val schema: StructType = StructType(List(
    StructField("fname", StringType, nullable = true),
    StructField("lname", StringType, nullable = true),
    StructField("married", StringType, nullable = true)
  ))
  val inputDataFrame: DataFrame = spark
    .read
    .schema(schema)
    .option("header",false)
    .option("delimiter",",")
    .csv("hydrograph.engine.spark/testData/inputFiles/delimitedInputFile.txt")
  //inputDataFrame.show()
  def isBad(row:Row):Boolean={
    row.anyNull
  }
  val simplefun=udf(isBad(_:Row))
  val cols=struct(inputDataFrame.schema.fieldNames.map(e=> col(e)):_*)
  //      println(cols+"******************") //for debugging
  val analysedDataFrame: DataFrame = inputDataFrame.withColumn("isRowBad", simplefun(cols))
  analysedDataFrame.show
}

如果你遇到任何问题，请给我回电。我相信这个解决方案是合适的，因为您似乎在寻找使用dataframe的代码。
谢谢。