Skip to content

[SPARK-9067][SQL] Close reader in NewHadoopRDD early if there is no more data #7424

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
viirya wants to merge 11 commits into from
Closed

[SPARK-9067][SQL] Close reader in NewHadoopRDD early if there is no more data #7424

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
67569da
Close reader early if there is no more data.
viirya Jul 15, 2015
a305621
Merge remote-tracking branch 'upstream/master' into close_reader
viirya Jul 16, 2015
f429016
Release reader if we don't need it.
viirya Jul 16, 2015
216912f
Fix it.
viirya Jul 16, 2015
50ed729
Merge remote-tracking branch 'upstream/master' into close_reader
viirya Jul 20, 2015
e34d98e
For comments.
viirya Jul 20, 2015
3ceb755
For comments.
viirya Jul 20, 2015
5116cbe
Merge remote-tracking branch 'upstream/master' into close_reader
viirya Jul 22, 2015
e152182
For comments.
viirya Jul 22, 2015
3d20267
Merge remote-tracking branch 'upstream/master' into close_reader
viirya Jul 24, 2015
3ff64e5
For comments.
viirya Jul 24, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 22 additions & 13 deletions core/src/main/scala/org/apache/spark/rdd/NewHadoopRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,7 @@ class NewHadoopRDD[K, V](
configurable.setConf(conf)
case _ =>
}
val reader = format.createRecordReader(
private var reader = format.createRecordReader(
split.serializableHadoopSplit.value, hadoopAttemptContext)
reader.initialize(split.serializableHadoopSplit.value, hadoopAttemptContext)

Expand All @@ -141,6 +141,10 @@ class NewHadoopRDD[K, V](
override def hasNext: Boolean = {
if (!finished && !havePair) {
finished = !reader.nextKeyValue
if (finished) {
Copy link
Contributor

@rxin rxin Jul 20, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

would it make sense if we just call close here?

// Close reader and release it
Copy link
Contributor

@kayousterhout kayousterhout Jul 23, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you change this comment (and the one in SqlNewHadoopRdd) to say something like "Close and release the reader here; close() will also be called when the task completes, but for tasks that read from many files, it helps to release the resources early"? I'm just worried this could be removed later on if someone things it's redundant with the close() in task completion listener.

Copy link
Member Author

@viirya viirya Jul 24, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good suggestion. I added it.

close()
}
havePair = !finished
}
!finished
Expand All @@ -159,18 +163,23 @@ class NewHadoopRDD[K, V](

private def close() {
try {
reader.close()
if (bytesReadCallback.isDefined) {
inputMetrics.updateBytesRead()
} else if (split.serializableHadoopSplit.value.isInstanceOf[FileSplit] ||
split.serializableHadoopSplit.value.isInstanceOf[CombineFileSplit]) {
// If we can't get the bytes read from the FS stats, fall back to the split size,
// which may be inaccurate.
try {
inputMetrics.incBytesRead(split.serializableHadoopSplit.value.getLength)
} catch {
case e: java.io.IOException =>
logWarning("Unable to get input size to set InputMetrics for task", e)
if (reader != null) {
// Close reader and release it
reader.close()
reader = null

if (bytesReadCallback.isDefined) {
inputMetrics.updateBytesRead()
} else if (split.serializableHadoopSplit.value.isInstanceOf[FileSplit] ||
split.serializableHadoopSplit.value.isInstanceOf[CombineFileSplit]) {
// If we can't get the bytes read from the FS stats, fall back to the split size,
// which may be inaccurate.
try {
inputMetrics.incBytesRead(split.serializableHadoopSplit.value.getLength)
} catch {
case e: java.io.IOException =>
logWarning("Unable to get input size to set InputMetrics for task", e)
}
}
}
} catch {
Expand Down
34 changes: 21 additions & 13 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/SqlNewHadoopRDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,7 +147,7 @@ private[sql] class SqlNewHadoopRDD[K, V](
configurable.setConf(conf)
case _ =>
}
val reader = format.createRecordReader(
private var reader = format.createRecordReader(
split.serializableHadoopSplit.value, hadoopAttemptContext)
reader.initialize(split.serializableHadoopSplit.value, hadoopAttemptContext)

Expand All @@ -160,6 +160,10 @@ private[sql] class SqlNewHadoopRDD[K, V](
override def hasNext: Boolean = {
if (!finished && !havePair) {
finished = !reader.nextKeyValue
if (finished) {
// Close reader and release it
close()
}
havePair = !finished
}
!finished
Expand All @@ -178,18 +182,22 @@ private[sql] class SqlNewHadoopRDD[K, V](

private def close() {
try {
reader.close()
if (bytesReadCallback.isDefined) {
inputMetrics.updateBytesRead()
} else if (split.serializableHadoopSplit.value.isInstanceOf[FileSplit] ||
split.serializableHadoopSplit.value.isInstanceOf[CombineFileSplit]) {
// If we can't get the bytes read from the FS stats, fall back to the split size,
// which may be inaccurate.
try {
inputMetrics.incBytesRead(split.serializableHadoopSplit.value.getLength)
} catch {
case e: java.io.IOException =>
logWarning("Unable to get input size to set InputMetrics for task", e)
if (reader != null) {
reader.close()
reader = null

if (bytesReadCallback.isDefined) {
inputMetrics.updateBytesRead()
} else if (split.serializableHadoopSplit.value.isInstanceOf[FileSplit] ||
split.serializableHadoopSplit.value.isInstanceOf[CombineFileSplit]) {
// If we can't get the bytes read from the FS stats, fall back to the split size,
// which may be inaccurate.
try {
inputMetrics.incBytesRead(split.serializableHadoopSplit.value.getLength)
} catch {
case e: java.io.IOException =>
logWarning("Unable to get input size to set InputMetrics for task", e)
}
}
}
} catch {
Expand Down