恐怕这行不通，至少对于当前版本的 Spark 是行不通的。
我不是 Spark 提交者，但我进行了调查，这就是我的发现。我不确定这是 100% 正确的，但也许它对你有用（至少作为进一步调查的良好起点）
我深入研究了 Spark 代码，发现多行文件和标准文件之间存在很大差异：
-
将 multiline 设置为 false Spark 正在使用 TextInputJsonDataSource 读取此文件，在这里您可以在代码 Spark Source Code 中看到读取操作的样子：
override def readFile(
conf: Configuration,
file: PartitionedFile,
parser: JacksonParser,
schema: StructType): Iterator[InternalRow] = {
val linesReader = new HadoopFileLinesReader(file, parser.options.lineSeparatorInRead, conf)
Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ => linesReader.close()))
val textParser = parser.options.encoding
.map(enc => CreateJacksonParser.text(enc, _: JsonFactory, _: Text))
.getOrElse(CreateJacksonParser.text(_: JsonFactory, _: Text))
val safeParser = new FailureSafeParser[Text](
input => parser.parse(input, textParser, textToUTF8String),
parser.options.parseMode,
schema,
parser.options.columnNameOfCorruptRecord)
linesReader.flatMap(safeParser.parse)
}
在这里我们可以看到 Spark 正在逐行读取文件，然后调用 flatMap 来用解析器处理每一行，这样以后很容易找到格式错误的记录并为它们生成 _corrupt_record
当您将 multiline 选项设置为 true 时，Spark 将使用 MultiLineJsonDataSource（剧透 - 它以前称为 WholeFileJsonDataSource）。在这里你可以找到读取数据的功能：Spark source code
override def readFile(
conf: Configuration,
file: PartitionedFile,
parser: JacksonParser,
schema: StructType): Iterator[InternalRow] = {
def partitionedFileString(ignored: Any): UTF8String = {
Utils.tryWithResource {
CodecStreams.createInputStreamWithCloseResource(conf, new Path(new URI(file.filePath)))
} { inputStream =>
UTF8String.fromBytes(ByteStreams.toByteArray(inputStream))
}
}
val streamParser = parser.options.encoding
.map(enc => CreateJacksonParser.inputStream(enc, _: JsonFactory, _: InputStream))
.getOrElse(CreateJacksonParser.inputStream(_: JsonFactory, _: InputStream))
val safeParser = new FailureSafeParser[InputStream](
input => parser.parse[InputStream](input, streamParser, partitionedFileString),
parser.options.parseMode,
schema,
parser.options.columnNameOfCorruptRecord)
safeParser.parse(
CodecStreams.createInputStreamWithCloseResource(conf, new Path(new URI(file.filePath))))
}
现在让我们看一下 JsonParser 及其通用函数解析：Spark source code
def parse[T](
record: T,
createParser: (JsonFactory, T) => JsonParser,
recordLiteral: T => UTF8String): Iterable[InternalRow] = {
try {
Utils.tryWithResource(createParser(factory, record)) { parser =>
// a null first token is equivalent to testing for input.trim.isEmpty
// but it works on any token stream and not just strings
parser.nextToken() match {
case null => None
case _ => rootConverter.apply(parser) match {
case null => throw QueryExecutionErrors.rootConverterReturnNullError()
case rows => rows.toSeq
}
}
}
} catch {
case e: SparkUpgradeException => throw e
case e @ (_: RuntimeException | _: JsonProcessingException | _: MalformedInputException) =>
// JSON parser currently doesnt support partial results for corrupted records.
// For such records, all fields other than the field configured by
// `columnNameOfCorruptRecord` are set to `null`
throw BadRecordException(() => recordLiteral(record), () => None, e)
case e: CharConversionException if options.encoding.isEmpty =>
val msg =
"""JSON parser cannot handle a character in its input.
|Specifying encoding as an input option explicitly might help to resolve the issue.
|""".stripMargin + e.getMessage
val wrappedCharException = new CharConversionException(msg)
wrappedCharException.initCause(e)
throw BadRecordException(() => recordLiteral(record), () => None, wrappedCharException)
case PartialResultException(row, cause) =>
throw BadRecordException(
record = () => recordLiteral(record),
partialResult = () => Some(row),
cause)
}
}
在这里您可以看到 Json 没有生成 PartialResultException，但可能是这两个异常中的一个：JsonProcessingException |畸形输入异常
由于此代码抛出此异常：BadRecordException(() => recordLiteral(record), () => None, e) 其中 record = our stream = whole file。
此异常稍后由 FailureSafeParser 解释，它为您生成输出行，并且只是将数据复制到 _corrupt_record
总的来说，我试图在提交和 Jira 中找到信息，但我认为这个主题真是一团糟。我找到了初始提交，它使用此消息添加了此功能：
[SPARK-18352][SQL] Support parsing multiline json files
## What changes were proposed in this pull request?
If a new option `wholeFile` is set to `true` the JSON reader will parse each file (instead of a single line) as a value. This is done with Jackson streaming and it should be capable of parsing very large documents, assuming the row will fit in memory.
Because the file is not buffered in memory the corrupt record handling is also slightly different when `wholeFile` is enabled: the corrupt column will contain the filename instead of the literal JSON if there is a parsing failure. It would be easy to extend this to add the parser location (line, column and byte offsets) to the output if desired.
“如果存在解析失败，损坏的列将包含文件名而不是文字 JSON”- 看起来后来有所更改（实际上您在此列中有文字 Json），但我认为一般方法是相同的。
所以回到问题：“我想知道这是预期的行为还是只是一个错误！” - 我认为这不是一个错误，也不是预期的行为，而是 Jackson 解析器最初实现方式的结果，目前我们必须忍受这个