Github user ho3rexqj commented on a diff in the pull request:
https://github.com/apache/spark/pull/20183#discussion_r161135870
--- Diff:
core/src/main/scala/org/apache/spark/broadcast/TorrentBroadcast.scala ---
@@ -206,37 +206,51 @@ private[spark] class TorrentBroadcast[T:
ClassTag](obj: T, id: Long)
private def readBroadcastBlock(): T = Utils.tryOrIOException {
TorrentBroadcast.synchronized {
- setConf(SparkEnv.get.conf)
- val blockManager = SparkEnv.get.blockManager
- 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()
- val blocks = readBlocks()
- logInfo("Reading broadcast variable " + id + " took" +
Utils.getUsedTimeMs(startTimeMs))
-
- try {
- val obj = TorrentBroadcast.unBlockifyObject[T](
- blocks.map(_.toInputStream()), SparkEnv.get.serializer,
compressionCodec)
- // Store the merged copy in BlockManager so other tasks on
this executor don't
- // need to re-fetch it.
- val storageLevel = StorageLevel.MEMORY_AND_DISK
- if (!blockManager.putSingle(broadcastId, obj, storageLevel,
tellMaster = false)) {
- throw new SparkException(s"Failed to store $broadcastId in
BlockManager")
+ val broadcastCache = SparkEnv.get.broadcastManager.cachedValues
+
+
Option(broadcastCache.get(broadcastId)).map(_.asInstanceOf[T]).getOrElse({
+ setConf(SparkEnv.get.conf)
--- End diff --
No, sorry - the cache update takes place within that block. With the
exception of those blocks (lines 220-222 and lines 244-246), yes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
