You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2023/09/04 21:37:44 UTC
[spark] branch branch-3.5 updated: [SPARK-44940][SQL][3.5] Improve performance of JSON parsing when "spark.sql.json.enablePartialResults" is enabled
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new 0dea7db3660 [SPARK-44940][SQL][3.5] Improve performance of JSON parsing when "spark.sql.json.enablePartialResults" is enabled
0dea7db3660 is described below
commit 0dea7db3660b9db7bbe075c31712e7119bfd1af7
Author: Ivan Sadikov <iv...@databricks.com>
AuthorDate: Mon Sep 4 14:37:32 2023 -0700
[SPARK-44940][SQL][3.5] Improve performance of JSON parsing when "spark.sql.json.enablePartialResults" is enabled
### What changes were proposed in this pull request?
Backport of https://github.com/apache/spark/pull/42667 to branch-3.5.
The PR improves JSON parsing when `spark.sql.json.enablePartialResults` is enabled:
- Fixes the issue when using nested arrays `ClassCastException: org.apache.spark.sql.catalyst.util.GenericArrayData cannot be cast to org.apache.spark.sql.catalyst.InternalRow`
- Improves parsing of the nested struct fields, e.g. `{"a1": "AAA", "a2": [{"f1": "", "f2": ""}], "a3": "id1", "a4": "XXX"}` used to be parsed as `|AAA|NULL |NULL|NULL|` and now is parsed as `|AAA|[{NULL, }]|id1|XXX|`.
- Improves performance of nested JSON parsing. The initial implementation would throw too many exceptions when multiple nested fields failed to parse. When the config is disabled, it is not a problem because the entire record is marked as NULL.
The internal benchmarks show the performance improvement from slowdown of over 160% to an improvement of 7-8% compared to the master branch when the flag is enabled. I will create a follow-up ticket to add a benchmark for this regression.
### Why are the changes needed?
Fixes some corner cases in JSON parsing and improves performance when `spark.sql.json.enablePartialResults` is enabled.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
I added tests to verify nested structs, maps, and arrays can be parsed without affecting the subsequent fields in the JSON. I also updated the existing tests when `spark.sql.json.enablePartialResults` is enabled because we parse more data now.
I added a benchmark to check performance.
Before the change (master, https://github.com/apache/spark/commit/a45a3a3d60cb97b107a177ad16bfe36372bc3e9b):
```
[info] OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
[info] Intel(R) Xeon(R) Platinum 8375C CPU 2.90GHz
[info] Partial JSON results: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
[info] ------------------------------------------------------------------------------------------------------------------------
[info] parse invalid JSON 9537 9820 452 0.0 953651.6 1.0X
```
After the change (this PR):
```
OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
Intel(R) Xeon(R) Platinum 8375C CPU 2.90GHz
Partial JSON results: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
parse invalid JSON 3100 3106 6 0.0 309967.6 1.0X
```
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #42790 from sadikovi/SPARK-44940-3.5.
Authored-by: Ivan Sadikov <iv...@databricks.com>
Signed-off-by: Dongjoon Hyun <dh...@apple.com>
---
.../spark/sql/catalyst/json/JacksonParser.scala | 41 ++++-
.../sql/catalyst/util/BadRecordException.scala | 64 +++++++-
.../spark/sql/errors/QueryExecutionErrors.scala | 12 +-
sql/core/benchmarks/JsonBenchmark-results.txt | 152 +++++++++---------
.../org/apache/spark/sql/JsonFunctionsSuite.scala | 20 ++-
.../execution/datasources/json/JsonBenchmark.scala | 28 ++++
.../sql/execution/datasources/json/JsonSuite.scala | 170 ++++++++++++++++++++-
7 files changed, 393 insertions(+), 94 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
index 388edb9024c..f14f70532e6 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala
@@ -420,17 +420,17 @@ class JacksonParser(
case VALUE_STRING if parser.getTextLength < 1 && allowEmptyString =>
dataType match {
case FloatType | DoubleType | TimestampType | DateType =>
- throw QueryExecutionErrors.emptyJsonFieldValueError(dataType)
+ throw EmptyJsonFieldValueException(dataType)
case _ => null
}
case VALUE_STRING if parser.getTextLength < 1 =>
- throw QueryExecutionErrors.emptyJsonFieldValueError(dataType)
+ throw EmptyJsonFieldValueException(dataType)
case token =>
// We cannot parse this token based on the given data type. So, we throw a
// RuntimeException and this exception will be caught by `parse` method.
- throw QueryExecutionErrors.cannotParseJSONFieldError(parser, token, dataType)
+ throw CannotParseJSONFieldException(parser.getCurrentName, parser.getText, token, dataType)
}
/**
@@ -459,6 +459,11 @@ class JacksonParser(
bitmask(index) = false
} catch {
case e: SparkUpgradeException => throw e
+ case err: PartialValueException if enablePartialResults =>
+ badRecordException = badRecordException.orElse(Some(err.cause))
+ row.update(index, err.partialResult)
+ skipRow = structFilters.skipRow(row, index)
+ bitmask(index) = false
case NonFatal(e) if isRoot || enablePartialResults =>
badRecordException = badRecordException.orElse(Some(e))
parser.skipChildren()
@@ -508,7 +513,7 @@ class JacksonParser(
if (badRecordException.isEmpty) {
mapData
} else {
- throw PartialResultException(InternalRow(mapData), badRecordException.get)
+ throw PartialMapDataResultException(mapData, badRecordException.get)
}
}
@@ -543,10 +548,21 @@ class JacksonParser(
throw PartialResultArrayException(arrayData.toArray[InternalRow](schema),
badRecordException.get)
} else {
- throw PartialResultException(InternalRow(arrayData), badRecordException.get)
+ throw PartialArrayDataResultException(arrayData, badRecordException.get)
}
}
+ /**
+ * Converts the non-stacktrace exceptions to user-friendly QueryExecutionErrors.
+ */
+ private def convertCauseForPartialResult(err: Throwable): Throwable = err match {
+ case CannotParseJSONFieldException(fieldName, fieldValue, jsonType, dataType) =>
+ QueryExecutionErrors.cannotParseJSONFieldError(fieldName, fieldValue, jsonType, dataType)
+ case EmptyJsonFieldValueException(dataType) =>
+ QueryExecutionErrors.emptyJsonFieldValueError(dataType)
+ case _ => err
+ }
+
/**
* Parse the JSON input to the set of [[InternalRow]]s.
*
@@ -589,12 +605,25 @@ class JacksonParser(
throw BadRecordException(
record = () => recordLiteral(record),
partialResults = () => Array(row),
- cause)
+ convertCauseForPartialResult(cause))
case PartialResultArrayException(rows, cause) =>
throw BadRecordException(
record = () => recordLiteral(record),
partialResults = () => rows,
cause)
+ // These exceptions should never be thrown outside of JacksonParser.
+ // They are used for the control flow in the parser. We add them here for completeness
+ // since they also indicate a bad record.
+ case PartialArrayDataResultException(arrayData, cause) =>
+ throw BadRecordException(
+ record = () => recordLiteral(record),
+ partialResults = () => Array(InternalRow(arrayData)),
+ convertCauseForPartialResult(cause))
+ case PartialMapDataResultException(mapData, cause) =>
+ throw BadRecordException(
+ record = () => recordLiteral(record),
+ partialResults = () => Array(InternalRow(mapData)),
+ convertCauseForPartialResult(cause))
}
}
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/BadRecordException.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/BadRecordException.scala
index e1223a71f74..65a56c1064e 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/BadRecordException.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/BadRecordException.scala
@@ -17,18 +17,44 @@
package org.apache.spark.sql.catalyst.util
+import com.fasterxml.jackson.core.JsonToken
+
import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.types.DataType
import org.apache.spark.unsafe.types.UTF8String
+abstract class PartialValueException(val cause: Throwable) extends Exception(cause) {
+ def partialResult: Serializable
+ override def getStackTrace(): Array[StackTraceElement] = cause.getStackTrace()
+ override def fillInStackTrace(): Throwable = this
+}
+
/**
- * Exception thrown when the underlying parser returns a partial result of parsing.
+ * Exception thrown when the underlying parser returns a partial result of parsing an object/row.
* @param partialResult the partial result of parsing a bad record.
* @param cause the actual exception about why the parser cannot return full result.
*/
case class PartialResultException(
- partialResult: InternalRow,
- cause: Throwable)
- extends Exception(cause)
+ override val partialResult: InternalRow,
+ override val cause: Throwable) extends PartialValueException(cause)
+
+/**
+ * Exception thrown when the underlying parser returns a partial array result.
+ * @param partialResult the partial array result.
+ * @param cause the actual exception about why the parser cannot return full result.
+ */
+case class PartialArrayDataResultException(
+ override val partialResult: ArrayData,
+ override val cause: Throwable) extends PartialValueException(cause)
+
+/**
+ * Exception thrown when the underlying parser returns a partial map result.
+ * @param partialResult the partial map result.
+ * @param cause the actual exception about why the parser cannot return full result.
+ */
+case class PartialMapDataResultException(
+ override val partialResult: MapData,
+ override val cause: Throwable) extends PartialValueException(cause)
/**
* Exception thrown when the underlying parser returns partial result list of parsing.
@@ -56,3 +82,33 @@ case class BadRecordException(
* Exception thrown when the underlying parser parses a JSON array as a struct.
*/
case class JsonArraysAsStructsException() extends RuntimeException()
+
+/**
+ * Exception thrown when the underlying parser can not parses a String as a datatype.
+ */
+case class StringAsDataTypeException(
+ fieldName: String,
+ fieldValue: String,
+ dataType: DataType) extends RuntimeException()
+
+/**
+ * No-stacktrace equivalent of `QueryExecutionErrors.cannotParseJSONFieldError`.
+ * Used for code control flow in the parser without overhead of creating a full exception.
+ */
+case class CannotParseJSONFieldException(
+ fieldName: String,
+ fieldValue: String,
+ jsonType: JsonToken,
+ dataType: DataType) extends RuntimeException() {
+ override def getStackTrace(): Array[StackTraceElement] = new Array[StackTraceElement](0)
+ override def fillInStackTrace(): Throwable = this
+}
+
+/**
+ * No-stacktrace equivalent of `QueryExecutionErrors.emptyJsonFieldValueError`.
+ * Used for code control flow in the parser without overhead of creating a full exception.
+ */
+case class EmptyJsonFieldValueException(dataType: DataType) extends RuntimeException() {
+ override def getStackTrace(): Array[StackTraceElement] = new Array[StackTraceElement](0)
+ override def fillInStackTrace(): Throwable = this
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
index 45986e42348..f4968cd0057 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
@@ -1329,11 +1329,19 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
def cannotParseJSONFieldError(parser: JsonParser, jsonType: JsonToken, dataType: DataType)
: SparkRuntimeException = {
+ cannotParseJSONFieldError(parser.getCurrentName, parser.getText, jsonType, dataType)
+ }
+
+ def cannotParseJSONFieldError(
+ fieldName: String,
+ fieldValue: String,
+ jsonType: JsonToken,
+ dataType: DataType): SparkRuntimeException = {
new SparkRuntimeException(
errorClass = "CANNOT_PARSE_JSON_FIELD",
messageParameters = Map(
- "fieldName" -> toSQLValue(parser.getCurrentName, StringType),
- "fieldValue" -> parser.getText,
+ "fieldName" -> toSQLValue(fieldName, StringType),
+ "fieldValue" -> fieldValue,
"jsonType" -> jsonType.toString(),
"dataType" -> toSQLType(dataType)))
}
diff --git a/sql/core/benchmarks/JsonBenchmark-results.txt b/sql/core/benchmarks/JsonBenchmark-results.txt
index 55f66f7bb24..e53c7801141 100644
--- a/sql/core/benchmarks/JsonBenchmark-results.txt
+++ b/sql/core/benchmarks/JsonBenchmark-results.txt
@@ -3,121 +3,125 @@ Benchmark for performance of JSON parsing
================================================================================================
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 1.8.0_362-b09 on Linux 5.15.0-1037-azure
-Intel(R) Xeon(R) CPU E5-2673 v3 @ 2.40GHz
+OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
JSON schema inferring: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 3720 3843 121 1.3 743.9 1.0X
-UTF-8 is set 5412 5455 45 0.9 1082.4 0.7X
+No encoding 2084 2134 46 2.4 416.8 1.0X
+UTF-8 is set 3077 3093 14 1.6 615.3 0.7X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 1.8.0_362-b09 on Linux 5.15.0-1037-azure
-Intel(R) Xeon(R) CPU E5-2673 v3 @ 2.40GHz
+OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
count a short column: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 3234 3254 33 1.5 646.7 1.0X
-UTF-8 is set 4847 4868 21 1.0 969.5 0.7X
+No encoding 2854 2863 8 1.8 570.8 1.0X
+UTF-8 is set 4066 4066 1 1.2 813.1 0.7X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 1.8.0_362-b09 on Linux 5.15.0-1037-azure
-Intel(R) Xeon(R) CPU E5-2673 v3 @ 2.40GHz
+OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
count a wide column: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 5702 5794 101 0.2 5702.1 1.0X
-UTF-8 is set 9526 9607 73 0.1 9526.1 0.6X
+No encoding 3348 3368 26 0.3 3347.8 1.0X
+UTF-8 is set 5215 5239 22 0.2 5214.7 0.6X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 1.8.0_362-b09 on Linux 5.15.0-1037-azure
-Intel(R) Xeon(R) CPU E5-2673 v3 @ 2.40GHz
+OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
select wide row: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-No encoding 18318 18448 199 0.0 366367.7 1.0X
-UTF-8 is set 19791 19887 99 0.0 395817.1 0.9X
+No encoding 11046 11102 54 0.0 220928.4 1.0X
+UTF-8 is set 12135 12181 54 0.0 242697.4 0.9X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 1.8.0_362-b09 on Linux 5.15.0-1037-azure
-Intel(R) Xeon(R) CPU E5-2673 v3 @ 2.40GHz
+OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
Select a subset of 10 columns: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Select 10 columns 2531 2570 51 0.4 2531.3 1.0X
-Select 1 column 1867 1882 16 0.5 1867.0 1.4X
+Select 10 columns 2486 2488 2 0.4 2486.5 1.0X
+Select 1 column 1505 1506 2 0.7 1504.6 1.7X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 1.8.0_362-b09 on Linux 5.15.0-1037-azure
-Intel(R) Xeon(R) CPU E5-2673 v3 @ 2.40GHz
+OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
creation of JSON parser per line: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Short column without encoding 868 875 7 1.2 868.4 1.0X
-Short column with UTF-8 1151 1163 11 0.9 1150.9 0.8X
-Wide column without encoding 12063 12299 205 0.1 12063.0 0.1X
-Wide column with UTF-8 16095 16136 51 0.1 16095.3 0.1X
+Short column without encoding 888 889 3 1.1 887.6 1.0X
+Short column with UTF-8 1134 1136 2 0.9 1134.3 0.8X
+Wide column without encoding 8012 8056 51 0.1 8012.4 0.1X
+Wide column with UTF-8 9830 9844 22 0.1 9829.7 0.1X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 1.8.0_362-b09 on Linux 5.15.0-1037-azure
-Intel(R) Xeon(R) CPU E5-2673 v3 @ 2.40GHz
+OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
JSON functions: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Text read 165 170 4 6.1 164.7 1.0X
-from_json 2339 2386 77 0.4 2338.9 0.1X
-json_tuple 2667 2730 55 0.4 2667.3 0.1X
-get_json_object 2627 2659 32 0.4 2627.1 0.1X
+Text read 85 87 2 11.7 85.4 1.0X
+from_json 1706 1711 4 0.6 1706.4 0.1X
+json_tuple 1528 1534 7 0.7 1528.2 0.1X
+get_json_object 1275 1286 17 0.8 1275.0 0.1X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 1.8.0_362-b09 on Linux 5.15.0-1037-azure
-Intel(R) Xeon(R) CPU E5-2673 v3 @ 2.40GHz
+OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
Dataset of json strings: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Text read 700 715 20 7.1 140.1 1.0X
-schema inferring 3144 3166 20 1.6 628.7 0.2X
-parsing 3261 3271 9 1.5 652.1 0.2X
+Text read 369 370 1 13.6 73.8 1.0X
+schema inferring 1880 1883 4 2.7 376.0 0.2X
+parsing 3731 3737 8 1.3 746.1 0.1X
Preparing data for benchmarking ...
-OpenJDK 64-Bit Server VM 1.8.0_362-b09 on Linux 5.15.0-1037-azure
-Intel(R) Xeon(R) CPU E5-2673 v3 @ 2.40GHz
+OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
Json files in the per-line mode: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Text read 1096 1105 12 4.6 219.1 1.0X
-Schema inferring 3818 3830 16 1.3 763.6 0.3X
-Parsing without charset 4107 4137 32 1.2 821.4 0.3X
-Parsing with UTF-8 5717 5763 41 0.9 1143.3 0.2X
+Text read 553 579 32 9.0 110.6 1.0X
+Schema inferring 2195 2196 2 2.3 439.0 0.3X
+Parsing without charset 4272 4274 3 1.2 854.3 0.1X
+Parsing with UTF-8 5459 5464 6 0.9 1091.7 0.1X
-OpenJDK 64-Bit Server VM 1.8.0_362-b09 on Linux 5.15.0-1037-azure
-Intel(R) Xeon(R) CPU E5-2673 v3 @ 2.40GHz
+OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
Write dates and timestamps: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-Create a dataset of timestamps 199 202 3 5.0 198.9 1.0X
-to_json(timestamp) 1458 1487 26 0.7 1458.0 0.1X
-write timestamps to files 1232 1262 26 0.8 1232.5 0.2X
-Create a dataset of dates 231 237 5 4.3 230.8 0.9X
-to_json(date) 956 966 10 1.0 956.5 0.2X
-write dates to files 785 793 10 1.3 785.4 0.3X
+Create a dataset of timestamps 102 112 13 9.8 101.9 1.0X
+to_json(timestamp) 840 841 1 1.2 839.6 0.1X
+write timestamps to files 692 696 4 1.4 692.0 0.1X
+Create a dataset of dates 120 121 1 8.4 119.7 0.9X
+to_json(date) 589 591 2 1.7 589.3 0.2X
+write dates to files 442 445 2 2.3 442.3 0.2X
-OpenJDK 64-Bit Server VM 1.8.0_362-b09 on Linux 5.15.0-1037-azure
-Intel(R) Xeon(R) CPU E5-2673 v3 @ 2.40GHz
+OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
Read dates and timestamps: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
-----------------------------------------------------------------------------------------------------------------------------------------------------
-read timestamp text from files 294 300 6 3.4 293.8 1.0X
-read timestamps from files 3254 3283 49 0.3 3254.0 0.1X
-infer timestamps from files 8390 8528 165 0.1 8389.8 0.0X
-read date text from files 269 276 7 3.7 269.3 1.1X
-read date from files 1178 1192 13 0.8 1177.8 0.2X
-timestamp strings 406 418 15 2.5 406.2 0.7X
-parse timestamps from Dataset[String] 3700 3713 16 0.3 3699.5 0.1X
-infer timestamps from Dataset[String] 8604 8647 65 0.1 8604.0 0.0X
-date strings 464 479 14 2.2 463.7 0.6X
-parse dates from Dataset[String] 1528 1538 10 0.7 1527.7 0.2X
-from_json(timestamp) 5402 5429 26 0.2 5401.8 0.1X
-from_json(date) 2948 2966 17 0.3 2947.5 0.1X
-infer error timestamps from Dataset[String] with default format 2358 2434 70 0.4 2357.6 0.1X
-infer error timestamps from Dataset[String] with user-provided format 2363 2390 36 0.4 2362.9 0.1X
-infer error timestamps from Dataset[String] with legacy format 2248 2287 35 0.4 2248.3 0.1X
+read timestamp text from files 143 145 4 7.0 142.6 1.0X
+read timestamps from files 2449 2469 17 0.4 2448.6 0.1X
+infer timestamps from files 5579 5596 15 0.2 5578.8 0.0X
+read date text from files 132 139 7 7.6 131.7 1.1X
+read date from files 1017 1020 2 1.0 1017.5 0.1X
+timestamp strings 227 230 3 4.4 227.2 0.6X
+parse timestamps from Dataset[String] 2827 2830 3 0.4 2826.5 0.1X
+infer timestamps from Dataset[String] 6001 6008 6 0.2 6001.2 0.0X
+date strings 259 261 2 3.9 259.0 0.6X
+parse dates from Dataset[String] 1382 1387 4 0.7 1382.3 0.1X
+from_json(timestamp) 3557 3561 7 0.3 3556.8 0.0X
+from_json(date) 2146 2148 2 0.5 2146.4 0.1X
+infer error timestamps from Dataset[String] with default format 1989 1993 4 0.5 1989.3 0.1X
+infer error timestamps from Dataset[String] with user-provided format 1922 1925 3 0.5 1922.1 0.1X
+infer error timestamps from Dataset[String] with legacy format 1919 1923 4 0.5 1919.1 0.1X
-OpenJDK 64-Bit Server VM 1.8.0_362-b09 on Linux 5.15.0-1037-azure
-Intel(R) Xeon(R) CPU E5-2673 v3 @ 2.40GHz
+OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
Filters pushdown: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
-w/o filters 22544 22661 109 0.0 225436.4 1.0X
-pushdown disabled 21045 21213 188 0.0 210452.6 1.1X
-w/ filters 893 904 10 0.1 8931.8 25.2X
-
+w/o filters 14387 14399 12 0.0 143871.9 1.0X
+pushdown disabled 13891 13899 7 0.0 138912.3 1.0X
+w/ filters 782 784 2 0.1 7820.5 18.4X
+OpenJDK 64-Bit Server VM 1.8.0_292-8u292-b10-0ubuntu1~18.04-b10 on Linux 5.4.0-1045-aws
+Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz
+Partial JSON results: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+------------------------------------------------------------------------------------------------------------------------
+parse invalid JSON 3100 3106 6 0.0 309967.6 1.0X
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
index 187fab75f63..a76e102fe91 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
@@ -1021,16 +1021,23 @@ class JsonFunctionsSuite extends QueryTest with SharedSparkSession {
.add("c2", ArrayType(new StructType().add("c3", LongType).add("c4", StringType)))
val df1 = Seq("""{"c2": [19], "c1": 123456}""").toDF("c0")
checkAnswer(df1.select(from_json($"c0", st)), Row(Row(123456, null)))
+
val df2 = Seq("""{"data": {"c2": [19], "c1": 123456}}""").toDF("c0")
- checkAnswer(df2.select(from_json($"c0", new StructType().add("data", st))), Row(Row(null)))
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "true") {
+ checkAnswer(
+ df2.select(from_json($"c0", new StructType().add("data", st))),
+ Row(Row(Row(123456, null)))
+ )
+ }
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "false") {
+ checkAnswer(df2.select(from_json($"c0", new StructType().add("data", st))), Row(Row(null)))
+ }
+ val df3 = Seq("""[{"c2": [19], "c1": 123456}]""").toDF("c0")
withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "true") {
- val df3 = Seq("""[{"c2": [19], "c1": 123456}]""").toDF("c0")
checkAnswer(df3.select(from_json($"c0", ArrayType(st))), Row(Array(Row(123456, null))))
}
-
withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "false") {
- val df3 = Seq("""[{"c2": [19], "c1": 123456}]""").toDF("c0")
checkAnswer(df3.select(from_json($"c0", ArrayType(st))), Row(null))
}
@@ -1119,14 +1126,13 @@ class JsonFunctionsSuite extends QueryTest with SharedSparkSession {
)
)
- // Value "a" cannot be parsed as an integer,
- // the error cascades to "c2", thus making its value null.
+ // Value "a" cannot be parsed as an integer, c2 value is null.
val df = Seq("""[{"c1": [{"c2": ["a"]}]}]""").toDF("c0")
withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "true") {
checkAnswer(
df.select(from_json($"c0", ArrayType(st))),
- Row(Array(Row(null)))
+ Row(Array(Row(Seq(Row(null)))))
)
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala
index c522378a65d..5b86543648f 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonBenchmark.scala
@@ -542,6 +542,33 @@ object JsonBenchmark extends SqlBasedBenchmark {
}
}
+ private def partialResultBenchmark(rowsNum: Int, numIters: Int): Unit = {
+ val benchmark = new Benchmark("Partial JSON results", rowsNum, output = output)
+ val colsNum = 1000
+
+ val fields = Seq.tabulate(colsNum)(i => StructField(s"col$i", IntegerType))
+ val schema = StructType(fields)
+
+ def data: Dataset[String] = {
+ spark.range(0, rowsNum, 1, 1).mapPartitions { iter =>
+ iter.map { i =>
+ (0 until colsNum).map { j =>
+ // Only the last column has an integer value.
+ if (j < colsNum - 1) s""""col${i}":"foo_${j}"""" else s""""col${i}":${j}"""
+ }.mkString("{", ", ", "}")
+ }
+ }.select($"value").as[String]
+ }
+
+ benchmark.addCase("parse invalid JSON", numIters) { _ =>
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "true") {
+ spark.read.schema(schema).json(data).noop()
+ }
+ }
+
+ benchmark.run()
+ }
+
override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
val numIters = 3
runBenchmark("Benchmark for performance of JSON parsing") {
@@ -558,6 +585,7 @@ object JsonBenchmark extends SqlBasedBenchmark {
// Benchmark pushdown filters that refer to top-level columns.
// TODO (SPARK-32325): Add benchmarks for filters with nested column attributes.
filtersPushdownBenchmark(rowsNum = 100 * 1000, numIters)
+ partialResultBenchmark(rowsNum = 10000, numIters)
}
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
index 95468a1f1d7..11779286ec2 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
@@ -3437,7 +3437,7 @@ abstract class JsonSuite
if (enablePartialResults) {
checkAnswer(
df,
- Seq(Row(null, Row(1)), Row(Row(2, null), Row(2)))
+ Seq(Row(Row(1, null), Row(1)), Row(Row(2, null), Row(2)))
)
} else {
checkAnswer(
@@ -3450,6 +3450,174 @@ abstract class JsonSuite
}
}
+ test("SPARK-44940: fully parse the record except f1 if partial results are enabled") {
+ withTempPath { path =>
+ Seq(
+ """{"a1": "AAA", "a2": [{"f1": "", "f2": ""}], "a3": "id1", "a4": "XXX"}""",
+ """{"a1": "BBB", "a2": [{"f1": 12, "f2": ""}], "a3": "id2", "a4": "YYY"}""").toDF()
+ .repartition(1)
+ .write.text(path.getAbsolutePath)
+
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "true") {
+ val df = spark.read.json(path.getAbsolutePath)
+ checkAnswer(
+ df,
+ Seq(
+ Row("AAA", Seq(Row(null, "")), "id1", "XXX"),
+ Row("BBB", Seq(Row(12, "")), "id2", "YYY")
+ )
+ )
+ }
+
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "false") {
+ val df = spark.read.json(path.getAbsolutePath)
+ checkAnswer(
+ df,
+ Seq(
+ Row("AAA", null, null, null),
+ Row("BBB", Seq(Row(12, "")), "id2", "YYY")
+ )
+ )
+ }
+ }
+ }
+
+ test("SPARK-44940: fully parse primitive map if partial results are enabled") {
+ withTempPath { path =>
+ Seq(
+ """{"a1": "AAA", "a2": {"f1": "", "f2": ""}, "a3": "id1"}""",
+ """{"a1": "BBB", "a2": {"f1": 12, "f2": ""}, "a3": "id2"}""").toDF()
+ .repartition(1)
+ .write.text(path.getAbsolutePath)
+
+ val schema = "a1 string, a2 map<string, int>, a3 string"
+
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "true") {
+ val df = spark.read.schema(schema).json(path.getAbsolutePath)
+ // Although the keys match the string type and some values match the integer type, because
+ // some of the values do not match the type, we mark the entire map as null.
+ checkAnswer(
+ df,
+ Seq(
+ Row("AAA", null, "id1"),
+ Row("BBB", null, "id2")
+ )
+ )
+ }
+
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "false") {
+ val df = spark.read.schema(schema).json(path.getAbsolutePath)
+ checkAnswer(
+ df,
+ Seq(
+ Row("AAA", null, null),
+ Row("BBB", null, null)
+ )
+ )
+ }
+ }
+ }
+
+ test("SPARK-44940: fully parse map of structs if partial results are enabled") {
+ withTempPath { path =>
+ Seq(
+ """{"a1": "AAA", "a2": {"key": {"f1": "", "f2": ""}}, "a3": "id1"}""",
+ """{"a1": "BBB", "a2": {"key": {"f1": 12, "f2": ""}}, "a3": "id2"}""").toDF()
+ .repartition(1)
+ .write.text(path.getAbsolutePath)
+
+ val schema = "a1 string, a2 map<string, struct<f1: int, f2: string>>, a3 string"
+
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "true") {
+ val df = spark.read.schema(schema).json(path.getAbsolutePath)
+ checkAnswer(
+ df,
+ Seq(
+ Row("AAA", Map("key" -> Row(null, "")), "id1"),
+ Row("BBB", Map("key" -> Row(12, "")), "id2")
+ )
+ )
+ }
+
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "false") {
+ val df = spark.read.schema(schema).json(path.getAbsolutePath)
+ checkAnswer(
+ df,
+ Seq(
+ Row("AAA", null, null),
+ Row("BBB", Map("key" -> Row(12, "")), "id2")
+ )
+ )
+ }
+ }
+ }
+
+ test("SPARK-44940: fully parse primitive arrays if partial results are enabled") {
+ withTempPath { path =>
+ Seq(
+ """{"a1": "AAA", "a2": {"f1": [""]}, "a3": "id1", "a4": "XXX"}""",
+ """{"a1": "BBB", "a2": {"f1": [12]}, "a3": "id2", "a4": "YYY"}""").toDF()
+ .repartition(1)
+ .write.text(path.getAbsolutePath)
+
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "true") {
+ val df = spark.read.json(path.getAbsolutePath)
+ checkAnswer(
+ df,
+ Seq(
+ Row("AAA", Row(null), "id1", "XXX"),
+ Row("BBB", Row(Seq(12)), "id2", "YYY")
+ )
+ )
+ }
+
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "false") {
+ val df = spark.read.json(path.getAbsolutePath)
+ checkAnswer(
+ df,
+ Seq(
+ Row("AAA", null, null, null),
+ Row("BBB", Row(Seq(12)), "id2", "YYY")
+ )
+ )
+ }
+ }
+ }
+
+ test("SPARK-44940: fully parse array of arrays if partial results are enabled") {
+ withTempPath { path =>
+ Seq(
+ """{"a1": "AAA", "a2": [[12, ""], [""]], "a3": "id1", "a4": "XXX"}""",
+ """{"a1": "BBB", "a2": [[12, 34], [""]], "a3": "id2", "a4": "YYY"}""").toDF()
+ .repartition(1)
+ .write.text(path.getAbsolutePath)
+
+ // We cannot parse `array<array<int>>` type because one of the inner arrays contains a
+ // mismatched type.
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "true") {
+ val df = spark.read.json(path.getAbsolutePath)
+ checkAnswer(
+ df,
+ Seq(
+ Row("AAA", null, "id1", "XXX"),
+ Row("BBB", null, "id2", "YYY")
+ )
+ )
+ }
+
+ withSQLConf(SQLConf.JSON_ENABLE_PARTIAL_RESULTS.key -> "false") {
+ val df = spark.read.json(path.getAbsolutePath)
+ checkAnswer(
+ df,
+ Seq(
+ Row("AAA", null, "id1", "XXX"),
+ Row("BBB", null, "id2", "YYY")
+ )
+ )
+ }
+ }
+ }
+
test("SPARK-40667: validate JSON Options") {
assert(JSONOptions.getAllOptions.size == 28)
// Please add validation on any new Json options here
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org