Skip to content
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

Jump to bottom

fix: only require barrier mode if more than one partition is trained on #1912

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

fix: only require barrier mode if more than one partition is trained on #1912

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c057e34
only require barrier mode if more than one partition is trained on
eisber Apr 11, 2023
ba4856b
Merge branch 'master' into marcozo/barrier-fix
mhamilton723 Apr 24, 2023
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
4 changes: 2 additions & 2 deletions vw/src/main/scala/com/microsoft/azure/synapse/ml/vw/VowpalWabbitBaseLearner.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,9 +179,9 @@ trait VowpalWabbitBaseLearner extends VowpalWabbitBase {
// schedule multiple mapPartitions in
val localInitialModel = if (isDefined(initialModel)) Some(getInitialModel) else None

// dispatch to exectuors and collect the model of the first partition (everybody has the same at the end anyway)
// dispatch to executors and collect the model of the first partition (everybody has the same at the end anyway)
// important to trigger collect() here so that the spanning tree is still up
if (getUseBarrierExecutionMode)
if (getUseBarrierExecutionMode && df.rdd.getNumPartitions > 1)
df.rdd.barrier().mapPartitions(inputRows => trainIteration(inputRows, localInitialModel)).collect().toSeq
else
df.mapPartitions(inputRows => trainIteration(inputRows, localInitialModel))(encoder).collect().toSeq
Expand Down
Loading