[GitHub] spark pull request #19464: Spark 22233
GitHub user liutang123 opened a pull request: https://github.com/apache/spark/pull/19464 Spark 22233 ## What changes were proposed in this pull request? add spark.hadoop.filterOutEmptySplit confituration to allow user to filter out empty split in HadoopRDD. You can merge this pull request into a Git repository by running: $ git pull https://github.com/liutang123/spark SPARK-22233 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/spark/pull/19464.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #19464 commit 2317bfdf18fc1a7b21cd43e0ec12f5e957fb1895 Author: liutang123 Date: 2017-06-21T04:27:42Z Merge pull request #1 from apache/master 20170521 pull request commit e3f993959fabdb80b966a42bf40d1cb5c6b44d95 Author: liulijia Date: 2017-09-28T06:12:04Z Merge branch 'master' of https://github.com/apache/spark commit 8f57d43b6bf127fc67e3e391d851efae3a859206 Author: liulijia Date: 2017-10-10T02:16:18Z Merge branch 'master' of https://github.com/apache/spark commit 3610f78837f4a5623f6d47b9feab1e565ed6 Author: liulijia Date: 2017-10-10T10:19:29Z allow user to filter empty split in HadoopRDD --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r143984294
--- Diff: core/src/main/scala/org/apache/spark/rdd/NewHadoopRDD.scala ---
@@ -122,7 +122,10 @@ class NewHadoopRDD[K, V](
case _ =>
}
val jobContext = new JobContextImpl(_conf, jobId)
-val rawSplits = inputFormat.getSplits(jobContext).toArray
+var rawSplits =
inputFormat.getSplits(jobContext).toArray(Array.empty[InputSplit])
+if
(sparkContext.getConf.getBoolean("spark.hadoop.filterOutEmptySplit", false)) {
+ rawSplits = rawSplits.filter(_.getLength>0)
--- End diff --
Space around operator.
You should filter before making an array.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144031167
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,16 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("spark.hadoop.filterOutEmptySplit") {
+val sf = new SparkConf()
+
sf.setAppName("test").setMaster("local").set("spark.hadoop.filterOutEmptySplit",
"true")
+sc = new SparkContext(sf)
+val emptyRDD = sc.parallelize(Array.empty[Tuple2[String, String]], 1)
+emptyRDD.saveAsHadoopFile[TextOutputFormat[String,
String]](tempDir.getPath + "/output")
+assert(new File(tempDir.getPath + "/output/part-0").exists() ===
true)
+
+val hadoopRDD = sc.textFile(tempDir.getPath + "/output/part-0")
+assert(hadoopRDD.partitions.length === 0)
--- End diff --
You should recycle the resources you required in the test case.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144030646
--- Diff: core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala ---
@@ -196,7 +196,10 @@ class HadoopRDD[K, V](
// add the credentials here as this can be called before SparkContext
initialized
SparkHadoopUtil.get.addCredentials(jobConf)
val inputFormat = getInputFormat(jobConf)
-val inputSplits = inputFormat.getSplits(jobConf, minPartitions)
+var inputSplits = inputFormat.getSplits(jobConf, minPartitions)
+if
(sparkContext.getConf.getBoolean("spark.hadoop.filterOutEmptySplit", false)) {
+ inputSplits = inputSplits.filter(_.getLength>0)
--- End diff --
nit: extra space around operator.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144031728
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,16 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("spark.hadoop.filterOutEmptySplit") {
+val sf = new SparkConf()
+
sf.setAppName("test").setMaster("local").set("spark.hadoop.filterOutEmptySplit",
"true")
+sc = new SparkContext(sf)
+val emptyRDD = sc.parallelize(Array.empty[Tuple2[String, String]], 1)
+emptyRDD.saveAsHadoopFile[TextOutputFormat[String,
String]](tempDir.getPath + "/output")
+assert(new File(tempDir.getPath + "/output/part-0").exists() ===
true)
+
+val hadoopRDD = sc.textFile(tempDir.getPath + "/output/part-0")
--- End diff --
We should also add the following test cases:
1. Ensure that if no split is empty, we don't lose any splits;
2. Ensure that if part of the splits are empty, we remove the splits
correctly.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user jiangxb1987 commented on a diff in the pull request: https://github.com/apache/spark/pull/19464#discussion_r144029852 --- Diff: docs/configuration.md --- @@ -1211,6 +1211,14 @@ Apart from these, the following properties are also available, and may be useful data may need to be rewritten to pre-existing output directories during checkpoint recovery. +spark.hadoop.filterOutEmptySplit --- End diff -- We should add the config to `internal/config`. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144030461
--- Diff: core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala ---
@@ -196,7 +196,10 @@ class HadoopRDD[K, V](
// add the credentials here as this can be called before SparkContext
initialized
SparkHadoopUtil.get.addCredentials(jobConf)
val inputFormat = getInputFormat(jobConf)
-val inputSplits = inputFormat.getSplits(jobConf, minPartitions)
+var inputSplits = inputFormat.getSplits(jobConf, minPartitions)
--- End diff --
How about:
```
val inputSplits = if (..) {
inputFormat.getSplits(jobConf, minPartitions).filter(_.getLength > 0)
} else {
inputFormat.getSplits(jobConf, minPartitions)
}
```
We should alway try to not use `var`.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user jerryshao commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144181321
--- Diff: core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala ---
@@ -196,7 +196,10 @@ class HadoopRDD[K, V](
// add the credentials here as this can be called before SparkContext
initialized
SparkHadoopUtil.get.addCredentials(jobConf)
val inputFormat = getInputFormat(jobConf)
-val inputSplits = inputFormat.getSplits(jobConf, minPartitions)
+var inputSplits = inputFormat.getSplits(jobConf, minPartitions)
+if
(sparkContext.getConf.getBoolean("spark.hadoop.filterOutEmptySplit", false)) {
--- End diff --
I would suggest not to use the name started by "spark.hadoop", this kind of
configurations will be treated as Hadoop configuration and set into Hadoop
`Configuration`, it might be better to choose another name.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144197159
--- Diff: core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala ---
@@ -196,7 +196,10 @@ class HadoopRDD[K, V](
// add the credentials here as this can be called before SparkContext
initialized
SparkHadoopUtil.get.addCredentials(jobConf)
val inputFormat = getInputFormat(jobConf)
-val inputSplits = inputFormat.getSplits(jobConf, minPartitions)
+var inputSplits = inputFormat.getSplits(jobConf, minPartitions)
+if
(sparkContext.getConf.getBoolean("spark.hadoop.filterOutEmptySplit", false)) {
--- End diff --
I'd use `spark.files` prefix, taken after `spark.files.ignoreCorruptFiles`,
`spark.files.maxPartitionBytes` and `spark.files.openCostInBytes`.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user liutang123 commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144235417
--- Diff: core/src/main/scala/org/apache/spark/rdd/NewHadoopRDD.scala ---
@@ -122,7 +122,10 @@ class NewHadoopRDD[K, V](
case _ =>
}
val jobContext = new JobContextImpl(_conf, jobId)
-val rawSplits = inputFormat.getSplits(jobContext).toArray
+var rawSplits =
inputFormat.getSplits(jobContext).toArray(Array.empty[InputSplit])
+if
(sparkContext.getConf.getBoolean("spark.hadoop.filterOutEmptySplit", false)) {
+ rawSplits = rawSplits.filter(_.getLength>0)
--- End diff --
Is there any one use empty file to do something ?
for example:
sc.textFile("/somepath/*").mapPartitions()
setting this flag to true by default may change the behavior of user's
application.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user liutang123 commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144235787
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,16 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("spark.hadoop.filterOutEmptySplit") {
+val sf = new SparkConf()
+
sf.setAppName("test").setMaster("local").set("spark.hadoop.filterOutEmptySplit",
"true")
+sc = new SparkContext(sf)
+val emptyRDD = sc.parallelize(Array.empty[Tuple2[String, String]], 1)
+emptyRDD.saveAsHadoopFile[TextOutputFormat[String,
String]](tempDir.getPath + "/output")
+assert(new File(tempDir.getPath + "/output/part-0").exists() ===
true)
+
+val hadoopRDD = sc.textFile(tempDir.getPath + "/output/part-0")
+assert(hadoopRDD.partitions.length === 0)
--- End diff --
The resources will be recycled by default in the afterEach function.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144311831
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,65 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("allow user to filter out empty split (old Hadoop API)") {
+val sf = new SparkConf()
+sf.setAppName("test").setMaster("local").set(FILTER_OUT_EMPTY_SPLIT,
true)
+sc = new SparkContext(sf)
+
+// Ensure that if all of the splits are empty, we remove the splits
correctly
+val emptyRDD = sc.parallelize(Array.empty[Tuple2[String, String]], 1)
+emptyRDD.saveAsHadoopFile[TextOutputFormat[String,
String]](tempDir.getPath + "/output")
--- End diff --
don't hardcode the path separator, use `new File(tempDir, output)`.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user srowen commented on a diff in the pull request: https://github.com/apache/spark/pull/19464#discussion_r144320027 --- Diff: docs/configuration.md --- @@ -1192,6 +1192,14 @@ Apart from these, the following properties are also available, and may be useful +spark.files.filterOutEmptySplit --- End diff -- I don't think I'd document this. It should be just a safety valve flag --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144319668
--- Diff: core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala ---
@@ -196,7 +196,11 @@ class HadoopRDD[K, V](
// add the credentials here as this can be called before SparkContext
initialized
SparkHadoopUtil.get.addCredentials(jobConf)
val inputFormat = getInputFormat(jobConf)
-val inputSplits = inputFormat.getSplits(jobConf, minPartitions)
+val inputSplits = if
(sparkContext.getConf.get(FILTER_OUT_EMPTY_SPLIT)) {
--- End diff --
You can avoid duplicating `inputFormat.getSplits(jobConf, minPartitions)`
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144319836
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,65 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("allow user to filter out empty split (old Hadoop API)") {
+val sf = new SparkConf()
--- End diff --
sf -> conf. You can fix it above too.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144319503
--- Diff:
core/src/main/scala/org/apache/spark/internal/config/package.scala ---
@@ -270,6 +270,15 @@ package object config {
.longConf
.createWithDefault(4 * 1024 * 1024)
+ private [spark] val FILTER_OUT_EMPTY_SPLIT =
ConfigBuilder("spark.files.filterOutEmptySplit")
--- End diff --
Nit: no space after private
This doc is much too verbose for a flag. Just say, "If true, methods like
that use HadoopRDD and NewHadoopRDD such as SparkContext.textFiles will not
create a partition for input splits that are empty."
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user cloud-fan commented on a diff in the pull request: https://github.com/apache/spark/pull/19464#discussion_r144324123 --- Diff: docs/configuration.md --- @@ -1192,6 +1192,14 @@ Apart from these, the following properties are also available, and may be useful +spark.files.filterOutEmptySplit --- End diff -- yea we can make this conf an internal conf. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144330429
--- Diff:
core/src/main/scala/org/apache/spark/internal/config/package.scala ---
@@ -270,6 +270,15 @@ package object config {
.longConf
.createWithDefault(4 * 1024 * 1024)
+ private [spark] val FILTER_OUT_EMPTY_SPLIT =
ConfigBuilder("spark.files.filterOutEmptySplit")
--- End diff --
nit: how about `ignoreEmptySplits` to be matched with `ignoreCorruptFiles`?
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144353093
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,54 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("spark.files.ignoreEmptySplits work correctly (old Hadoop API)") {
+val conf = new SparkConf()
+conf.setAppName("test").setMaster("local").set(IGNORE_EMPTY_SPLITS,
true)
+sc = new SparkContext(conf)
+
+def testIgnoreEmptySplits(data: Array[Tuple2[String, String]],
numSlices: Int,
+ outputSuffix: Int, checkPart: String,
partitionLength: Int): Unit = {
+ val dataRDD = sc.parallelize(data, numSlices)
+ val output = new File(tempDir, "output" + outputSuffix)
+ dataRDD.saveAsHadoopFile[TextOutputFormat[String,
String]](output.getPath)
+ assert(new File(output, checkPart).exists() === true)
+ val hadoopRDD = sc.textFile(new File(output, "part-*").getPath)
+ assert(hadoopRDD.partitions.length === partitionLength)
+}
+
+// Ensure that if all of the splits are empty, we remove the splits
correctly
+testIgnoreEmptySplits(Array.empty[Tuple2[String, String]], 1, 0,
"part-0", 0)
--- End diff --
I'd call it with named arguments, for example,
```scala
testIgnoreEmptySplits(
Array.empty[Tuple2[String, String]],
numSlices = 1,
outputSuffix = 0,
checkPart = "part-0",
expectedPartitionNum = 0)
```
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144357089
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,83 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("spark.files.ignoreEmptySplits work correctly (old Hadoop API)") {
+val conf = new SparkConf()
+conf.setAppName("test").setMaster("local").set(IGNORE_EMPTY_SPLITS,
true)
+sc = new SparkContext(conf)
+
+def testIgnoreEmptySplits(data: Array[Tuple2[String, String]],
numSlices: Int,
+ outputSuffix: Int, checkPart: String, expectedPartitionNum: Int):
Unit = {
+ val dataRDD = sc.parallelize(data, numSlices)
+ val output = new File(tempDir, "output" + outputSuffix)
+ dataRDD.saveAsHadoopFile[TextOutputFormat[String,
String]](output.getPath)
+ assert(new File(output, checkPart).exists() === true)
+ val hadoopRDD = sc.textFile(new File(output, "part-*").getPath)
+ assert(hadoopRDD.partitions.length === expectedPartitionNum)
+}
--- End diff --
Could we maybe do this something like ... as below? (not tested)
```scala
def testIgnoreEmptySplits(
data: Array[Tuple2[String, String]],
actualPartitionNum: Int,
expectedName: String,
expectedPartitionNum: Int): Unit = {
val output = new File(tempDir, "output")
sc.parallelize(data, actualPartitionNum)
.saveAsHadoopFile[TextOutputFormat[String,
String]](output.getAbsolutePath)
assert(new File(output, expectedPart).exists())
val hadoopRDD = sc.textFile(new File(output, "part-*").getAbsolutePath)
assert(hadoopRDD.partitions.length === expectedPartitionNum)
}
...
testIgnoreEmptySplits(
data = Array.empty[Tuple2[String, String]],
actualPartitionNum = 1,
expectedName = "part-0",
expectedPartitionNum = 0)
```
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144357376
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,83 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("spark.files.ignoreEmptySplits work correctly (old Hadoop API)") {
+val conf = new SparkConf()
+conf.setAppName("test").setMaster("local").set(IGNORE_EMPTY_SPLITS,
true)
+sc = new SparkContext(conf)
+
+def testIgnoreEmptySplits(data: Array[Tuple2[String, String]],
numSlices: Int,
+ outputSuffix: Int, checkPart: String, expectedPartitionNum: Int):
Unit = {
+ val dataRDD = sc.parallelize(data, numSlices)
+ val output = new File(tempDir, "output" + outputSuffix)
+ dataRDD.saveAsHadoopFile[TextOutputFormat[String,
String]](output.getPath)
+ assert(new File(output, checkPart).exists() === true)
+ val hadoopRDD = sc.textFile(new File(output, "part-*").getPath)
+ assert(hadoopRDD.partitions.length === expectedPartitionNum)
+}
+
+// Ensure that if all of the splits are empty, we remove the splits
correctly
+testIgnoreEmptySplits(
+ data = Array.empty[Tuple2[String, String]],
+ numSlices = 1,
+ outputSuffix = 0,
+ checkPart = "part-0",
+ expectedPartitionNum = 0)
+
+// Ensure that if no split is empty, we don't lose any splits
+testIgnoreEmptySplits(
+ data = Array(("key1", "a"), ("key2", "a"), ("key3", "b")),
+ numSlices = 2,
+ outputSuffix = 1,
+ checkPart = "part-1",
+ expectedPartitionNum = 2)
+
+// Ensure that if part of the splits are empty, we remove the splits
correctly
+testIgnoreEmptySplits(
+ data = Array(("key1", "a"), ("key2", "a")),
+ numSlices = 5,
+ outputSuffix = 2,
+ checkPart = "part-4",
+ expectedPartitionNum = 2)
+ }
+
+ test("spark.files.ignoreEmptySplits work correctly (new Hadoop API)") {
+val conf = new SparkConf()
+conf.setAppName("test").setMaster("local").set(IGNORE_EMPTY_SPLITS,
true)
+sc = new SparkContext(conf)
+
+def testIgnoreEmptySplits(data: Array[Tuple2[String, String]],
numSlices: Int,
+ outputSuffix: Int, checkPart: String, expectedPartitionNum: Int):
Unit = {
+ val dataRDD = sc.parallelize(data, numSlices)
+ val output = new File(tempDir, "output" + outputSuffix)
+ dataRDD.saveAsNewAPIHadoopFile[NewTextOutputFormat[String,
String]](output.getPath)
+ assert(new File(output, checkPart).exists() === true)
+ val hadoopRDD = sc.textFile(new File(output, "part-r-*").getPath)
--- End diff --
I think we should _read_ it with new hadoop API to test `NewHadoopRDD` I
guess?
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144365104
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,83 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("spark.files.ignoreEmptySplits work correctly (old Hadoop API)") {
+val conf = new SparkConf()
+conf.setAppName("test").setMaster("local").set(IGNORE_EMPTY_SPLITS,
true)
+sc = new SparkContext(conf)
+
+def testIgnoreEmptySplits(data: Array[Tuple2[String, String]],
numSlices: Int,
+ outputSuffix: Int, checkPart: String, expectedPartitionNum: Int):
Unit = {
+ val dataRDD = sc.parallelize(data, numSlices)
+ val output = new File(tempDir, "output" + outputSuffix)
+ dataRDD.saveAsHadoopFile[TextOutputFormat[String,
String]](output.getPath)
+ assert(new File(output, checkPart).exists() === true)
+ val hadoopRDD = sc.textFile(new File(output, "part-*").getPath)
+ assert(hadoopRDD.partitions.length === expectedPartitionNum)
+}
--- End diff --
Actually, to me the previous tests were also okay to me as well ..
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144452771
--- Diff:
core/src/main/scala/org/apache/spark/internal/config/package.scala ---
@@ -270,6 +270,12 @@ package object config {
.longConf
.createWithDefault(4 * 1024 * 1024)
+ private[spark] val IGNORE_EMPTY_SPLITS =
ConfigBuilder("spark.files.ignoreEmptySplits")
+.doc("If true, methods like that use HadoopRDD and NewHadoopRDD such
as " +
--- End diff --
`like that` -> `that`
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user jerryshao commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144472244
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,86 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("spark.files.ignoreEmptySplits work correctly (old Hadoop API)") {
+val conf = new SparkConf()
+conf.setAppName("test").setMaster("local").set(IGNORE_EMPTY_SPLITS,
true)
+sc = new SparkContext(conf)
+
+def testIgnoreEmptySplits(data: Array[Tuple2[String, String]],
numSlices: Int,
--- End diff --
nit: one argument per line.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user jerryshao commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144472362
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,86 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("spark.files.ignoreEmptySplits work correctly (old Hadoop API)") {
+val conf = new SparkConf()
+conf.setAppName("test").setMaster("local").set(IGNORE_EMPTY_SPLITS,
true)
+sc = new SparkContext(conf)
+
+def testIgnoreEmptySplits(data: Array[Tuple2[String, String]],
numSlices: Int,
+ outputSuffix: Int, checkPart: String, expectedPartitionNum: Int):
Unit = {
+ val dataRDD = sc.parallelize(data, numSlices)
+ val output = new File(tempDir, "output" + outputSuffix)
+ dataRDD.saveAsHadoopFile[TextOutputFormat[String,
String]](output.getPath)
+ assert(new File(output, checkPart).exists() === true)
+ val hadoopRDD = sc.textFile(new File(output, "part-*").getPath)
+ assert(hadoopRDD.partitions.length === expectedPartitionNum)
+}
+
+// Ensure that if all of the splits are empty, we remove the splits
correctly
+testIgnoreEmptySplits(
+ data = Array.empty[Tuple2[String, String]],
+ numSlices = 1,
+ outputSuffix = 0,
+ checkPart = "part-0",
+ expectedPartitionNum = 0)
+
+// Ensure that if no split is empty, we don't lose any splits
+testIgnoreEmptySplits(
+ data = Array(("key1", "a"), ("key2", "a"), ("key3", "b")),
+ numSlices = 2,
+ outputSuffix = 1,
+ checkPart = "part-1",
+ expectedPartitionNum = 2)
+
+// Ensure that if part of the splits are empty, we remove the splits
correctly
+testIgnoreEmptySplits(
+ data = Array(("key1", "a"), ("key2", "a")),
+ numSlices = 5,
+ outputSuffix = 2,
+ checkPart = "part-4",
+ expectedPartitionNum = 2)
+ }
+
+ test("spark.files.ignoreEmptySplits work correctly (new Hadoop API)") {
+val conf = new SparkConf()
+conf.setAppName("test").setMaster("local").set(IGNORE_EMPTY_SPLITS,
true)
+sc = new SparkContext(conf)
+
+def testIgnoreEmptySplits(data: Array[Tuple2[String, String]],
numSlices: Int,
--- End diff --
ditto.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144604878
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,87 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("spark.files.ignoreEmptySplits work correctly (old Hadoop API)") {
+val conf = new SparkConf()
+conf.setAppName("test").setMaster("local").set(IGNORE_EMPTY_SPLITS,
true)
+sc = new SparkContext(conf)
+
+def testIgnoreEmptySplits(
+ data: Array[Tuple2[String, String]],
+ actualPartitionNum: Int,
+ expectedPart: String,
+ expectedPartitionNum: Int): Unit = {
+ val output = new File(tempDir, "output")
+ sc.parallelize(data, actualPartitionNum)
+.saveAsHadoopFile[TextOutputFormat[String, String]](output.getPath)
+ assert(new File(output, expectedPart).exists() === true)
--- End diff --
I don't think we need the `expectedPart` parameter, just
```
for (i <- 0 until actualPartitionNum) {
assert(new File(output, s"part-$i").exists() === true)
}
```
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144617011
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,87 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("spark.files.ignoreEmptySplits work correctly (old Hadoop API)") {
+val conf = new SparkConf()
+conf.setAppName("test").setMaster("local").set(IGNORE_EMPTY_SPLITS,
true)
+sc = new SparkContext(conf)
+
+def testIgnoreEmptySplits(
+ data: Array[Tuple2[String, String]],
+ actualPartitionNum: Int,
+ expectedPart: String,
+ expectedPartitionNum: Int): Unit = {
+ val output = new File(tempDir, "output")
+ sc.parallelize(data, actualPartitionNum)
+.saveAsHadoopFile[TextOutputFormat[String, String]](output.getPath)
+ assert(new File(output, expectedPart).exists() === true)
+ val hadoopRDD = sc.textFile(new File(output, "part-*").getPath)
+ assert(hadoopRDD.partitions.length === expectedPartitionNum)
+ Utils.deleteRecursively(output)
--- End diff --
Maybe:
```scala
try {
...
} finally {
Utils.deleteRecursively(output)
}
```
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144617310
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,87 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("spark.files.ignoreEmptySplits work correctly (old Hadoop API)") {
+val conf = new SparkConf()
+conf.setAppName("test").setMaster("local").set(IGNORE_EMPTY_SPLITS,
true)
+sc = new SparkContext(conf)
+
+def testIgnoreEmptySplits(
+ data: Array[Tuple2[String, String]],
+ actualPartitionNum: Int,
+ expectedPart: String,
+ expectedPartitionNum: Int): Unit = {
--- End diff --
Indentation..
```scala
def testIgnoreEmptySplits(
data: Array[Tuple2[String, String]],
...
expectedPartitionNum: Int): Unit = {
val output = new File(tempDir, "output")
...
```
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user liutang123 commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144683771
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,87 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("spark.files.ignoreEmptySplits work correctly (old Hadoop API)") {
+val conf = new SparkConf()
+conf.setAppName("test").setMaster("local").set(IGNORE_EMPTY_SPLITS,
true)
+sc = new SparkContext(conf)
+
+def testIgnoreEmptySplits(
+ data: Array[Tuple2[String, String]],
+ actualPartitionNum: Int,
+ expectedPart: String,
+ expectedPartitionNum: Int): Unit = {
+ val output = new File(tempDir, "output")
+ sc.parallelize(data, actualPartitionNum)
+.saveAsHadoopFile[TextOutputFormat[String, String]](output.getPath)
+ assert(new File(output, expectedPart).exists() === true)
+ val hadoopRDD = sc.textFile(new File(output, "part-*").getPath)
+ assert(hadoopRDD.partitions.length === expectedPartitionNum)
+ Utils.deleteRecursively(output)
--- End diff --
I think we don't need `try... finally` here. Because
`Utils.deleteRecursively(output)` just to ensure
the success of next invocation of the `testIgnoreEmptySplits`. When test
finished, wether be passed or not, the `tempDir` will be deleted in
`FileSuite.afterEach()`.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user HyukjinKwon commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144687101
--- Diff: core/src/test/scala/org/apache/spark/FileSuite.scala ---
@@ -510,4 +510,83 @@ class FileSuite extends SparkFunSuite with
LocalSparkContext {
}
}
+ test("spark.files.ignoreEmptySplits work correctly (old Hadoop API)") {
+val conf = new SparkConf()
+conf.setAppName("test").setMaster("local").set(IGNORE_EMPTY_SPLITS,
true)
+sc = new SparkContext(conf)
+
+def testIgnoreEmptySplits(
+data: Array[Tuple2[String, String]],
+actualPartitionNum: Int,
+expectedPartitionNum: Int): Unit = {
+ val output = new File(tempDir, "output")
+ sc.parallelize(data, actualPartitionNum)
+.saveAsHadoopFile[TextOutputFormat[String, String]](output.getPath)
+ for (i <- 0 until actualPartitionNum) {
+assert(new File(output, s"part-$i").exists() === true)
+ }
+ val hadoopRDD = sc.textFile(new File(output, "part-*").getPath)
+ assert(hadoopRDD.partitions.length === expectedPartitionNum)
+ Utils.deleteRecursively(output)
+}
+
+// Ensure that if all of the splits are empty, we remove the splits
correctly
+testIgnoreEmptySplits(
+ data = Array.empty[Tuple2[String, String]],
+ actualPartitionNum = 1,
+ expectedPartitionNum = 0)
+
+// Ensure that if no split is empty, we don't lose any splits
+testIgnoreEmptySplits(
+ data = Array(("key1", "a"), ("key2", "a"), ("key3", "b")),
+ actualPartitionNum = 2,
+ expectedPartitionNum = 2)
+
+// Ensure that if part of the splits are empty, we remove the splits
correctly
+testIgnoreEmptySplits(
+ data = Array(("key1", "a"), ("key2", "a")),
+ actualPartitionNum = 5,
+ expectedPartitionNum = 2)
+ }
+
+ test("spark.files.ignoreEmptySplits work correctly (new Hadoop API)") {
+val conf = new SparkConf()
+conf.setAppName("test").setMaster("local").set(IGNORE_EMPTY_SPLITS,
true)
+sc = new SparkContext(conf)
+
+def testIgnoreEmptySplits(
+data: Array[Tuple2[String, String]],
+actualPartitionNum: Int,
+expectedPartitionNum: Int): Unit = {
+ val output = new File(tempDir, "output")
+ sc.parallelize(data, actualPartitionNum)
+.saveAsNewAPIHadoopFile[NewTextOutputFormat[String,
String]](output.getPath)
+ for (i <- 0 until actualPartitionNum) {
+assert(new File(output, s"part-r-$i").exists() === true)
+ }
+ val hadoopRDD = sc.newAPIHadoopFile(new File(output,
"part-r-*").getPath,
+classOf[NewTextInputFormat], classOf[LongWritable], classOf[Text])
+.asInstanceOf[NewHadoopRDD[_, _]]
--- End diff --
nit:
```scala
val hadoopRDD = sc.newAPIHadoopFile(
new File(output, "part-r-*").getPath,
classOf[NewTextInputFormat],
classOf[LongWritable],
classOf[Text]).asInstanceOf[NewHadoopRDD[_, _]]
```
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user asfgit closed the pull request at: https://github.com/apache/spark/pull/19464 --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144745022
--- Diff:
core/src/main/scala/org/apache/spark/internal/config/package.scala ---
@@ -270,6 +270,12 @@ package object config {
.longConf
.createWithDefault(4 * 1024 * 1024)
+ private[spark] val IGNORE_EMPTY_SPLITS =
ConfigBuilder("spark.files.ignoreEmptySplits")
--- End diff --
This config should be made internal, and the name should be improved
because it's not about spark files.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #19464: [SPARK-22233] [core] Allow user to filter out emp...
Github user jiangxb1987 commented on a diff in the pull request:
https://github.com/apache/spark/pull/19464#discussion_r144745063
--- Diff:
core/src/main/scala/org/apache/spark/internal/config/package.scala ---
@@ -270,6 +270,12 @@ package object config {
.longConf
.createWithDefault(4 * 1024 * 1024)
+ private[spark] val IGNORE_EMPTY_SPLITS =
ConfigBuilder("spark.files.ignoreEmptySplits")
--- End diff --
I'll send a follow-up PR to fix this.
---
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
