-
Notifications
You must be signed in to change notification settings - Fork 29k
[SPARK-20386][Spark Core]modify the log info if the block exists on the slave already #17683
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.
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
Changes from 6 commits
312b90b
2d95652
22ef469
c14b1ad
b769754
81e8e84
567a9ae
f7e40b8
60ca9e0
664dfb8
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -497,11 +497,17 @@ private[spark] class BlockManagerInfo( | |
|
|
||
| updateLastSeenMs() | ||
|
|
||
| var blockExists = false | ||
| var originalMemSize: Long = 0 | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why pull out a var here? |
||
| var originalDiskSize: Long = 0 | ||
|
|
||
| if (_blocks.containsKey(blockId)) { | ||
| // The block exists on the slave already. | ||
| val blockStatus: BlockStatus = _blocks.get(blockId) | ||
| val originalLevel: StorageLevel = blockStatus.storageLevel | ||
| val originalMemSize: Long = blockStatus.memSize | ||
| originalMemSize = blockStatus.memSize | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. BLocks of code even farther down can reuse these new values and need similar changes to the log comment
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Ok,thanks for suggestion, I have modified it |
||
| originalDiskSize = blockStatus.diskSize | ||
| blockExists = true | ||
|
|
||
| if (originalLevel.useMemory) { | ||
| _remainingMem += originalMemSize | ||
|
|
@@ -516,19 +522,32 @@ private[spark] class BlockManagerInfo( | |
| * They can be both larger than 0, when a block is dropped from memory to disk. | ||
| * Therefore, a safe way to set BlockStatus is to set its info in accurate modes. */ | ||
| var blockStatus: BlockStatus = null | ||
|
|
||
| if (storageLevel.useMemory) { | ||
| blockStatus = BlockStatus(storageLevel, memSize = memSize, diskSize = 0) | ||
| _blocks.put(blockId, blockStatus) | ||
| _remainingMem -= memSize | ||
| logInfo("Added %s in memory on %s (size: %s, free: %s)".format( | ||
| blockId, blockManagerId.hostPort, Utils.bytesToString(memSize), | ||
| Utils.bytesToString(_remainingMem))) | ||
| if (blockExists) { | ||
| logInfo("Updated %s in memory on %s (current size: %s, original size %s, free: %s)".format( | ||
| blockId, blockManagerId.hostPort, Utils.bytesToString(memSize), | ||
| Utils.bytesToString(originalMemSize),Utils.bytesToString(_remainingMem))) | ||
| } else { | ||
| logInfo("Added %s in memory on %s (size: %s, free: %s)".format( | ||
| blockId, blockManagerId.hostPort, Utils.bytesToString(memSize), | ||
| Utils.bytesToString(_remainingMem))) | ||
| } | ||
| } | ||
| if (storageLevel.useDisk) { | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Doesn't the disk case need a similar treatment?
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Ok, I have modified the disk case, and add a "update" style log info, any other suggestions? thank you |
||
| blockStatus = BlockStatus(storageLevel, memSize = 0, diskSize = diskSize) | ||
| _blocks.put(blockId, blockStatus) | ||
| logInfo("Added %s on disk on %s (size: %s)".format( | ||
| blockId, blockManagerId.hostPort, Utils.bytesToString(diskSize))) | ||
| if (blockExists) { | ||
| logInfo("Updated %s on disk on %s (current size: %s, original size %s)".format( | ||
|
||
| blockId, blockManagerId.hostPort, Utils.bytesToString(diskSize), | ||
| Utils.bytesToString(originalDiskSize))) | ||
| } else { | ||
| logInfo("Added %s on disk on %s (size: %s)".format( | ||
| blockId, blockManagerId.hostPort, Utils.bytesToString(diskSize))) | ||
| } | ||
| } | ||
| if (!blockId.isBroadcast && blockStatus.isCached) { | ||
| _cachedBlocks += blockId | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Although you could just assign
val blockExists = _blocks.containsKey(blockId)once and reuse it instead of avar, it's minor. This looks good.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's a good idea, i have modified it, thank you