賞金將在 6 天后到期。此問題的答案有資格獲得 50聲望賞金。 Prateek Gautam正在從可靠的來源尋找答案：
我已經驗證了 spark 中多行 json 的作業，但它沒有按預期作業。我想知道這是預期的行為還是只是一個錯誤！因為如果我們將多行選項傳遞為真，那么我們希望 spark 將根陣列中的 json 物件視為不同的，而不是將整個檔案視為一個。
我正在使用 spark 2.7 版的 spark java 應用程式。我正在嘗試根據我的模式加載可能包含損壞記錄的多行 JSON 檔案。我在加載它時傳遞了一個模式，但問題是它拒絕整個檔案作為一個損壞的記錄，即使有一個 JSON 物件不滿足我提供的模式。
我的 Json 檔案看起來像-
[
{Json_object},
{Json_object},
{Json_object}
]
我為它手動創建了模式（StructType）并加載它 -
Dataset<Row> df = spark.read().option("multiline", "true").option("mode","PERMISSIVE").option("columnNameOfCorruptRecord","_corrupt_record").schema(schema).json("filepath");
問題是，即使一個 JSON 物件不遵循模式，例如，如果我的模式中的 attribute1 具有整數型別并且它是 json 物件之一的字串形式，那么 json 物件應該進入 corrupted_record，insted I'我得到類似的東西-
------------ --------------- ---------------
| attribute1 | attribute2 |_corrupt_record|
------------ --------------- ---------------
| null | null | [{|
| | | all_json_obj |
| | | ... |
| | | }] |
------------ --------------- ---------------
它在典型的單行 json 物件中作業得非常好，其中換行符 '\n' 被用作分隔符，沒有任何問題，并且結果理想。有人能告訴我我在這里想念什么嗎？
PS：問題不限于spark java，scala和python的行為也是一樣的。
uj5u.com熱心網友回復：
恐怕這行不通，至少對于當前版本的 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 源代碼
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 決議器最初實作方式的結果，目前我們必須忍受這個
uj5u.com熱心網友回復：
通過查看您的輸出，我將在此處復制：
------------ --------------- ---------------
| attribute1 | attribute2 |_corrupt_record|
------------ --------------- ---------------
| null | null | [{|
| | | all_json_obj |
| | | ... |
| | | }] |
------------ --------------- ---------------
如果您查看第一行和最后一行，您會看到 corrupt_records 是[{和}]。這讓我覺得可能那些{和}字符不應該在那里。是否有可能你的 json 檔案實際上是這樣的：
[{
{Json_object},
{Json_object},
{Json_object}
}]
如果是這種情況，那么{}最高級別[]方括號之間的花括號將使最高級別陣列看起來只包含 1 個物件，并且模式錯誤。如果是這種情況，您可以嘗試洗掉陣列方括號之間的花括號嗎？
只是為了給你一個功能示例，請考慮以下 json 檔案：
[
{
"id": 1,
"object": {
"val1": "thisValue",
"val2": "otherValue"
}
},
{
"id": 2,
"object": {
"val1": "hehe",
"val2": "test"
}
},
{
"id": 3,
"object": {
"val1": "yes",
"val2": "no"
}
}
]
使用以下命令在 spark-shell（spark 版本 2.4.5）中讀取該 json 檔案：
val df = spark.read.option("multiline", "true").json("test.json")
給我以下輸出：
scala> df.show(false)
--- -----------------------
|id |object |
--- -----------------------
|1 |[thisValue, otherValue]|
|2 |[hehe, test] |
|3 |[yes, no] |
--- -----------------------
scala> df.printSchema
root
|-- id: long (nullable = true)
|-- object: struct (nullable = true)
| |-- val1: string (nullable = true)
| |-- val2: string (nullable = true)
這只是一個小例子，可以為您提供一些功能。
但是一定要看看損壞的資料框中的那些[{和}]行！
希望能幫助到你 ：）
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/537493.html
標籤：阿帕奇火花pysparkapache-spark-sql火花爪哇