Skip to content

[SPARK-18827][Core] Fix cannot read broadcast on disk #16252

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
wangyum wants to merge 5 commits into from
Closed

[SPARK-18827][Core] Fix cannot read broadcast on disk #16252

Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
58acc06
Fix cannot read broadcast on disk
wangyum Dec 12, 2016
d964c54
Consistency with previously code
wangyum Dec 12, 2016
f004740
Make more robust
wangyum Dec 13, 2016
0903da8
Remove !unrolled.hasNext check.
wangyum Dec 14, 2016
c13edce
Restore !unrolled.hasNext check.
wangyum Dec 17, 2016
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
Prev Previous commit
Next Next commit
Make more robust
  • Loading branch information
@wangyum
wangyum committed Dec 13, 2016
commit f00474032dab2561e85d3c2fd7aad01d0dcacc8e
14 changes: 9 additions & 5 deletions core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -207,11 +207,15 @@ private[spark] class TorrentBroadcast[T: ClassTag](obj: T, id: Long)
TorrentBroadcast.synchronized {
setConf(SparkEnv.get.conf)
val blockManager = SparkEnv.get.blockManager
blockManager.getLocalValues(broadcastId).map(_.data.next()) match {
case Some(x) =>
releaseLock(broadcastId)
x.asInstanceOf[T]

blockManager.getLocalValues(broadcastId) match {
case Some(blockResult) =>
if (blockResult.data.hasNext) {
val x = blockResult.data.next().asInstanceOf[T]
releaseLock(broadcastId)
x
} else {
throw new SparkException(s"Failed to get locally stored broadcast data: $broadcastId")
}
case None =>
logInfo("Started reading broadcast variable " + id)
val startTimeMs = System.currentTimeMillis()
Expand Down