Spark在创建数据集时无法反序列化记录

这是我提出的解决方案（我正在使用Spark Structured Streaming）：

val stream = spark.readStream
.format("csv")
.schema(schema) // a StructType defined elsewhere
.option("mode", "PERMISSIVE")
.option("columnNameOfCorruptRecord", "corruptRecord")
.load(path)

// If debugging, check for any corrupted CSVs
if (log.isDebugEnabled) { // org.apache.spark.internal.Logging trait
import spark.implicits._
stream

.filter($"corruptRecord".isNotNull)
.withColumn("input_file", input_file_name)
.select($"input_file", $"corruptRecord")
.writeStream
.format("console")
.option("truncate", false)
.start()

}

val events = stream
.withColumn("event", lit("I"))
.withColumn("source", lit(sourceName))
.as[TradeRecord]
基本上，如果将Spark日志级别设置为Debug或更低，则会检查DataFrame是否存在损坏的记录，并将所有此类记录与其文件名一起打印出来。最终程序尝试将此DataFrame强制转换为强类型Dataset[TradeRecord]并失败。

根据user10465355的建议，您可以加载数据：

val events: DataFrame = ???
过滤

val mismatched = events.where($"deal".isNull)
添加文件名

import org.apache.spark.sql.functions.input_file_name

val tagged = mismatched.withColumn("_file_name", input_file_name)
（可选）添加块和块以及偏移量：

import org.apache.spark.sql.functions.{spark_partition_id, monotonically_increasing_id, shiftLeft, shiftRight

df
.withColumn("chunk", spark_partition_id())
.withColumn(

"offset",
monotonically_increasing_id - shiftLeft(shiftRight(monotonically_increasing_id, 33), 33))