Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/21056#discussion_r183227100
--- Diff:
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala
---
@@ -2128,38 +2128,77 @@ class JsonSuite extends QueryTest with
SharedSQLContext with TestJsonData {
}
}
- test("SPARK-23849: schema inferring touches less data if samplingRation
< 1.0") {
- val predefinedSample = Set[Int](2, 8, 15, 27, 30, 34, 35, 37, 44, 46,
+ val sampledTestData = (value: java.lang.Long) => {
+ val predefinedSample = Set[Long](2, 8, 15, 27, 30, 34, 35, 37, 44, 46,
57, 62, 68, 72)
- withTempPath { path =>
- val writer = Files.newBufferedWriter(Paths.get(path.getAbsolutePath),
- StandardCharsets.UTF_8, StandardOpenOption.CREATE_NEW)
- for (i <- 0 until 100) {
- if (predefinedSample.contains(i)) {
- writer.write(s"""{"f1":${i.toString}}""" + "\n")
- } else {
- writer.write(s"""{"f1":${(i.toDouble + 0.1).toString}}""" + "\n")
- }
- }
- writer.close()
-
- val ds = spark.read.option("samplingRatio",
0.1).json(path.getCanonicalPath)
- assert(ds.schema == new StructType().add("f1", LongType))
+ if (predefinedSample.contains(value)) {
+ s"""{"f1":${value.toString}}"""
+ } else {
+ s"""{"f1":${(value.toDouble + 0.1).toString}}"""
}
}
- test("SPARK-23849: usage of samplingRation while parsing of dataset of
strings") {
- val dstr = spark.sparkContext.parallelize(0 until 100, 1).map { i =>
- val predefinedSample = Set[Int](2, 8, 15, 27, 30, 34, 35, 37, 44, 46,
- 57, 62, 68, 72)
- if (predefinedSample.contains(i)) {
- s"""{"f1":${i.toString}}""" + "\n"
- } else {
- s"""{"f1":${(i.toDouble + 0.1).toString}}""" + "\n"
- }
- }.toDS()
- val ds = spark.read.option("samplingRatio", 0.1).json(dstr)
+ test("SPARK-23849: schema inferring touches less data if samplingRatio <
1.0") {
+ // Set default values for the DataSource parameters to make sure
+ // that whole test file is mapped to only one partition. This will
guarantee
+ // reliable sampling of the input file.
+ withSQLConf(
+ "spark.sql.files.maxPartitionBytes" -> (128 * 1024 * 1024).toString,
+ "spark.sql.files.openCostInBytes" -> (4 * 1024 * 1024).toString
+ )(withTempPath { path =>
+ val ds = spark.range(0, 100, 1, 1).map(sampledTestData)
+ ds.write.text(path.getAbsolutePath)
+ val readback = spark.read.option("samplingRatio",
0.1).json(path.getCanonicalPath)
+
+ assert(readback.schema == new StructType().add("f1", LongType))
+ })
+ }
+
+ test("SPARK-23849: usage of samplingRatio while parsing a dataset of
strings") {
+ val ds = spark.range(0, 100, 1, 1).map(sampledTestData)
+ val readback = spark.read.option("samplingRatio", 0.1).json(ds)
+
+ assert(readback.schema == new StructType().add("f1", LongType))
+ }
+
+ test("SPARK-23849: samplingRatio is out of the range (0, 1.0]") {
+ val ds = spark.range(0, 100, 1, 1).map(_.toString)
- assert(ds.schema == new StructType().add("f1", LongType))
+ val errorMsg0 = intercept[IllegalArgumentException] {
+ spark.read.option("samplingRatio", -1).json(ds)
+ }.getMessage
+ assert(errorMsg0.contains("samplingRatio (-1.0) should be greater than
0"))
+
+ val errorMsg1 = intercept[IllegalArgumentException] {
+ spark.read.option("samplingRatio", 0).json(ds)
+ }.getMessage
+ assert(errorMsg1.contains("samplingRatio (0.0) should be greater than
0"))
+
+ val sampled = spark.read.option("samplingRatio", 10).json(ds)
+ assert(sampled.count() == ds.count())
+ }
+
+ test("SPARK-23849: sampling files for schema inferring in the multiLine
mode") {
+ withTempDir { dir =>
+ Files.write(Paths.get(dir.getAbsolutePath, "0.json"),
"""{"a":"a"}""".getBytes,
--- End diff --
maybe `getBytes` with explicit encoding UTF-8.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
