-
Notifications
You must be signed in to change notification settings - Fork 142
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
[#615] improvement: Reduce task binary by removing 'partitionToServers' from RssShuffleHandle #637
Conversation
…m RssShuffleHandle Signed-off-by: jiafu zhang <jiafu.zhang@intel.com>
Codecov Report
@@ Coverage Diff @@
## master #637 +/- ##
============================================
+ Coverage 60.63% 62.78% +2.14%
Complexity 1845 1845
============================================
Files 228 215 -13
Lines 12702 10753 -1949
Branches 1062 1062
============================================
- Hits 7702 6751 -951
+ Misses 4592 3655 -937
+ Partials 408 347 -61
... and 15 files with indirect coverage changes 📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more |
…m RssShuffleHandle Signed-off-by: jiafu zhang <jiafu.zhang@intel.com>
PTAL @xianjingfeng first. I will review it later tonight. |
* Class for holding partition ID -> shuffle servers mapping. | ||
* It's to be broadcast to executors and referenced by shuffle tasks. | ||
*/ | ||
public class PartitionShuffleServerMap { |
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.
This class may store other information, such as RemoteStorageInfo
. I think we should use a better name. ShuffleHandlerInfo? I hope you have a better idea.
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.
sounds good.
client-spark/common/src/main/java/org/apache/spark/shuffle/RssShuffleHandle.java
Show resolved
Hide resolved
public int getShuffleId() { | ||
return shuffleId(); | ||
} | ||
|
||
public Set<ShuffleServerInfo> getShuffleServersForData() { |
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.
ditto
It's OK for me for the basic logic. You can deal with code style and ci first. @zjf2012 |
client-spark/common/src/main/java/org/apache/spark/shuffle/RssShuffleHandle.java
Show resolved
Hide resolved
import org.slf4j.Logger; | ||
import org.slf4j.LoggerFactory; | ||
|
||
import scala.reflect.ClassTag; |
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.
the import group order doesn't seem right.
Let's try spark way:
import java
import scala
import third_party
import uniffle
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.
let adjust it accordingly.
* @param partitionToServers | ||
* @return Broadcast variable registered for auto cleanup | ||
*/ | ||
public static Broadcast<PartitionShuffleServerMap> createPartShuffleServerMap( |
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.
I'm prefer to pass SparkContext
as a parameter here. it's much clear and we should pass SparkContext if we are going to add unit test for this method.
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.
I agree.
The code lgtm overall. some minor questions:
Do you have some data about the serialized task before and after, is there any slowdown when using broadcast and how many shuffle servers in your environments? And just to be safe, how much the size of broadcast is occupied when your 10000 partition test? Just to make sure it don't bring too much memory pressure to driver. |
For now, I don't have more servers. So, I only use two shuffle servers. Before my optimization, both map task and reduce task have more than 670KB binary size. After optimization, they reduce to less than 6KB. It's dramatic. Broadcast uses bittorrent-like way to distribute variable to each executor once. Executors can get some chunk of broadcast variable from other executors instead of all from driver. And task serialize/deserialize time drops a lot. So in theory, it has no way to slow down job. The size of the broadcast should be less than 670KB deduced from above statement. I'll try to capture it today. |
Will the broadcast be cleaned when the shuffle is removed? |
yes. it's registered for cleanup when sparkcontext creates a new broadcast variable. However, the cleaner may not clean it immediately just like below unregisterShuffle method. The cleaner gets chance to capture them (shuffle, broadcast...) only after GC happened. This behavior is good for this broadcast variable since it's small anyway. Will it cause problem in below code snippet? shuffle server may not get cleaned up quickly if there is no right GC in driver.
|
Ok |
identified two problems,
I need to serialize ShuffleHandleInfo to binary with spark closureSerialize before spark kryoserializing and broadcasting it. Working on it. |
…m RssShuffleHandle Signed-off-by: jiafu zhang <jiafu.zhang@intel.com>
…m RssShuffleHandle Signed-off-by: jiafu zhang <jiafu.zhang@intel.com>
Before my patch, "task + partition -> shuffle servers " size in binary is about 83.6KiB. Deserialized size is about 659.8KiB. 2023-02-22 13:36:13,581 INFO memory.MemoryStore: Block broadcast_2_piece0 stored as bytes in memory (estimated size 83.6 KiB, free 23.4 GiB) With my patch, the broadcasted variable size in binary is about 80.6KiB < 83.6KiB. Deserialized size is about 655KiB < 659.8KiB . They are as expected. 2023-02-23 12:43:20,015 INFO memory.MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 80.6 KiB, free 23.4 GiB) |
Please see vanished task deserialization time with patch from https://docs.google.com/document/d/1TZ-3Mgwj9j7n1mMyCrS3sskFv_uOtUtXl1oF9Y_oMOw/edit?usp=sharing. |
Could you modify the title and description? It seems that the title is too long, some words of title were added to description. |
client-spark/spark2/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java
Outdated
Show resolved
Hide resolved
client-spark/spark2/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java
Outdated
Show resolved
Hide resolved
client-spark/spark3/src/main/java/org/apache/spark/shuffle/RssShuffleManager.java
Outdated
Show resolved
Hide resolved
…m RssShuffleHandle Signed-off-by: jiafu zhang <jiafu.zhang@intel.com>
@advancedxy @xianjingfeng Do you have another suggestion? |
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.
LGTM
I'm not sure, this pr introduced some quite complex logic to broadcast shuffle handle info. If I was implementing this feature, I would just use Kryo by default, and in RSSShuffleManager indicating users to either |
Just make sure we are on the same page. Only registering RssShuffleHandle to kryo serializer doesn't help resolve this issue. Each task will still have more than 670KB size in binary. for 10000 partitions job. And each task will have quite noticeable deserialization time to repeatedly deserialize partition -> shuffle server mappings as shown in https://docs.google.com/document/d/1TZ-3Mgwj9j7n1mMyCrS3sskFv_uOtUtXl1oF9Y_oMOw/edit?usp=sharing. RssShuffleHandle is a field of ShuffleDependency which is serialized to task binary in below code in DAGScheduler. Without broadcast of ShuffleHandleInfo, it's hard to pull it out and avoid repeat of ShuffleHandleInfo. ` RDDCheckpointData.synchronized {
I think broadcast itself is quite simple and efficient. I don't see other better alternatives. |
Sorry I wasn't clear enough. I agree broadcast is the best way to go. I'm just not sure that we should broadcast binary rather than just broadcast the shuffle handle info simply. The only problem that we don't broadcast shuffle handle info directly is that once |
no worries. You guys can decide which option is better. I can revert back if you like. |
Thanks for your understanding. Let's take a quick vote here, cc @zuston @jerqi @xianjingfeng @kaijchen
|
// shuffle ID to ShuffleIdRef | ||
// ShuffleIdRef acts as strong reference to prevent cached ShuffleHandleInfo being GCed during shuffle | ||
// ShuffleIdRef will be removed when unregisterShuffle() | ||
private static Map<Integer, ShuffleIdRef> _globalShuffleIdRefMap = new ConcurrentHashMap<>(); |
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 seems that the underline does not conform to the name definition specification of java
. Right?
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.
yes. it's just a convention to denote special variable. If it's violate uniffle style, I can change it.
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.
done
// ShuffleIdRef will be removed when unregisterShuffle() | ||
private static Map<Integer, ShuffleIdRef> _globalShuffleIdRefMap = new ConcurrentHashMap<>(); | ||
// each shuffle has unique ID even for multiple concurrent running shuffles and jobs per application | ||
private static ThreadLocal<HandleInfoLocalCache> _localHandleInfoCache = |
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.
ditto.
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.
done
…m RssShuffleHandle Signed-off-by: jiafu zhang <jiafu.zhang@intel.com>
…m RssShuffleHandle revert to ShuffleHandleInfo without ser/deser Signed-off-by: jiafu zhang <jiafu.zhang@intel.com>
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.
LGTM! Thanks @zjf2012
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.
LGTM, except one minor comment.
client-spark/common/src/main/java/org/apache/spark/shuffle/RssSparkShuffleUtils.java
Show resolved
Hide resolved
thanks @zjf2012. I noticed there's no added UTs here, do you think it's possible to add new UTs or existing UTs are sufficient? |
You are welcome. I think current UT is good enough since my changes only wrap up some existing things. No complex logic involved. If I add some UT, it would be mockups for the broadcast, no much actual business logic. |
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.
LGTM
Thanks @zjf2012 |
…Servers' from RssShuffleHandle (apache#637) ### What changes were proposed in this pull request? move partition -> shuffle servers mapping from direct field of RssShuffleHandle to a broadcast variable to reduce task binary size. ### Why are the changes needed? to reduce task delay and task serialize/deserialize time by reduce task binary size ### Does this PR introduce any user-facing change? No. ### How was this patch tested? tested with 10000 partitions shuffle. Task binary size reduced from more than 670KB to less than 6KB. tested with multiple shuffle stages in same job to verify ShuffleHandleInfo cache logic
…Servers' from RssShuffleHandle (apache#637) ### What changes were proposed in this pull request? move partition -> shuffle servers mapping from direct field of RssShuffleHandle to a broadcast variable to reduce task binary size. ### Why are the changes needed? to reduce task delay and task serialize/deserialize time by reduce task binary size ### Does this PR introduce any user-facing change? No. ### How was this patch tested? tested with 10000 partitions shuffle. Task binary size reduced from more than 670KB to less than 6KB. tested with multiple shuffle stages in same job to verify ShuffleHandleInfo cache logic
I don't see the logic of destroy the broadcast var explicitly. Do we need to destroy it in the |
What changes were proposed in this pull request?
move partition -> shuffle servers mapping from direct field of RssShuffleHandle to a broadcast variable to reduce task binary size.
Why are the changes needed?
to reduce task delay and task serialize/deserialize time by reduce task binary size
Does this PR introduce any user-facing change?
No.
How was this patch tested?