Skip to content

[SPARK-39983][CORE][SQL] Do not cache unserialized broadcast relations on the driver #37413

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

[SPARK-39983][CORE][SQL] Do not cache unserialized broadcast relations on the driver #37413

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e6baa25
SPARK-39983 - do not cache unserialized broadcast relations on the dr…
alex-balikov Aug 5, 2022
75ab18e
comment fixes
alex-balikov Aug 5, 2022
cb2bfa0
Store WeakReference for serializedOnly broadcast variables
JoshRosen Aug 9, 2022
63db4ff
Empty-Commit
alex-balikov Aug 10, 2022
6a725a8
Empty-Commit
alex-balikov Aug 10, 2022
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
comment fixes
  • Loading branch information
@alex-balikov
alex-balikov committed Aug 5, 2022
commit 75ab18ee0e382b8117bf65fc9ef05190d4fdf01a
3 changes: 2 additions & 1 deletion 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 @@ -143,7 +143,8 @@ private[spark] class TorrentBroadcast[T: ClassTag](obj: T, id: Long, serializedO
// regressions if an internal broadcast is accessed on the driver, we store a soft
// reference to the broadcasted value:
_value = new SoftReference[T](value)
} else { // Store a copy of the broadcast variable in the driver so that tasks run on the driver
} else {
// Store a copy of the broadcast variable in the driver so that tasks run on the driver
// do not create a duplicate copy of the broadcast variable's value.
if (!blockManager.putSingle(broadcastId, value, MEMORY_AND_DISK, tellMaster = false)) {
throw new SparkException(s"Failed to store $broadcastId in BlockManager")
Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,7 @@ case class BroadcastExchangeExec(
val beforeBroadcast = System.nanoTime()
longMetric("buildTime") += NANOSECONDS.toMillis(beforeBroadcast - beforeBuild)

// SPARK-39983 - Broadcast the relation without caching the unserialized value.
// SPARK-39983 - Broadcast the relation without caching the unserialized object.
val broadcasted = sparkContext.broadcastInternal(relation, serializedOnly = true)
longMetric("broadcastTime") += NANOSECONDS.toMillis(
System.nanoTime() - beforeBroadcast)
Expand Down