[WIP] Use spark barrier mode to ensure all xgboost worker launched parallelly #5625
Conversation
|
is there any battle-tested cases of barrier execution in Spark? I didn't see enough ROI for such a fundamental change....(and I didn't see changes covering the original behavior that, e.g., stop the application after a single worker failed ) |
In barrier mode, if one worker failed, all other workers will be also killed and spark job failed.
The main benefit will be If we set spark cluster executor to be auto scale mode, then old approach won't work. If worker is not enough it won't trigger more executor allocation. But barrier mode will address it, it will trigger spark cluster to create more executors. What about we keep old behavior, but add an option for add new barrier mode ? |
you mean dynamic allocation? then just check whether users sets minExecutors in dynamic allocation no smaller than numWorkers
we never officially support such cases (due to some issues in rabit layer)...even we support, to work around these, add several lines to detect if static allocation, when launching a training, always check if it is possible to get enough resources given the configured number of executors if dynamic allocation, I think we can do similarly, (but I really don't think we should spend too much time in dynamic allocation...in Uber, dynamic allocation leads to many difficulties for tuning) |
This don't address issue I think. e.g., if one xgboost job occupy current executors resources, and started a new xgboost job on the same spark cluster, spark should dynamic allocate more executors rather than waiting task slots on current executors become idle. |
Could you please elaborate a bit on this? |
| @@ -548,9 +546,6 @@ object XGBoost extends Serializable { | |||
| // Train for every ${savingRound} rounds and save the partially completed booster | |||
| val tracker = startTracker(xgbExecParams.numWorkers, xgbExecParams.trackerConf) | |||
| val (booster, metrics) = try { | |||
| val parallelismTracker = new SparkParallelismTracker(sc, | |||
| xgbExecParams.timeoutRequestWorkers, | |||
There was a problem hiding this comment.
timeoutRequestWorkers can be removed from XGBoostExecutionParams.
| logger.info(s"Rabit returns with exit code $trackerReturnVal") | ||
| val (booster, metrics) = postTrackerReturnProcessing(trackerReturnVal, |
There was a problem hiding this comment.
L665-692 private def postTrackerReturnProcessing(... is no longer used.
There is a senario that use spark CrossValidation on Xgboost estimator, it will launch Xgboost jobs parallelly. If do not address this issue, deadlock may happen. And seemingly we have no way to workaround (such as add several lines to detect, it does not work, e.g., if two threads check resources and both get available response, then they start there own xgboost jobs parallely, then deadlock will still happen) And what's the issue you mentioned in "due to some issues in rabit layer" ? What do you think ? |
Current xgoost-4j-spark use
SparkParallelismTrackerto wait and check whether there's enough task slots. This approach has two issue:in dead lock.
In spark 2.4, barrier mode was introduced, which used to ensure a spark job stage launch all tasks parallelly. This address the issue above.
In this PR , I remove
SparkParallelismTrackerand update code to use spark barrier mode.Note: now we cannot set xgboost params
timeout_request_workersfor each individual job. We have to setspark.scheduler.barrier.maxConcurrentTasksCheck.intervalandspark.scheduler.barrier.maxConcurrentTasksCheck.maxFailuresin spark cluster config. The timeout to wait for enough workers will be equal tospark.scheduler.barrier.maxConcurrentTasksCheck.interval * spark.scheduler.barrier.maxConcurrentTasksCheck.maxFailuresWill add test soon but it is ready for first pass review.