Github user dhruve commented on a diff in the pull request:
https://github.com/apache/spark/pull/21601#discussion_r199597945
--- Diff:
core/src/main/scala/org/apache/spark/input/WholeTextFileInputFormat.scala ---
@@ -53,6 +53,19 @@ private[spark] class WholeTextFileInputFormat
val totalLen = files.map(file => if (file.isDirectory) 0L else
file.getLen).sum
val maxSplitSize = Math.ceil(totalLen * 1.0 /
(if (minPartitions == 0) 1 else minPartitions)).toLong
+
+ // For small files we need to ensure the min split size per node &
rack <= maxSplitSize
+ val config = context.getConfiguration
+ val minSplitSizePerNode =
config.getLong(CombineFileInputFormat.SPLIT_MINSIZE_PERNODE, 0L)
+ val minSplitSizePerRack =
config.getLong(CombineFileInputFormat.SPLIT_MINSIZE_PERRACK, 0L)
+
+ if (maxSplitSize < minSplitSizePerNode) {
+ super.setMinSplitSizeNode(maxSplitSize)
--- End diff --
AFAIU If we set these to `0L` unconditionally, every time there is left
over data which wasn't combined into a split, would result in its own split
because minSplitSizePerNode is `0L`.
This shouldn't be an issue for small no. of files. But if we have a large
no. of small files which result in a similar situation, we will end up having
more splits rather than combining these together to form lesser no. of splits.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
