lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-2901] Fixed the bug clustering jobs cannot running in parallel (#4178)

Browse Source
This commit is contained in:
xiarixiaoyao
2021-12-10 14:39:35 +08:00
committed by GitHub
parent ea154bcb5d
commit 456d74ce4e
2 changed files with 52 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java
View File

@@ -35,6 +35,7 @@ import org.apache.hudi.common.model.HoodieRecordPayload;
import org.apache.hudi.common.model.RewriteAvroPayload;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner;
import org.apache.hudi.common.util.FutureUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.collection.Pair;
@@ -88,16 +89,17 @@ public abstract class MultipleSparkJobExecutionStrategy<T extends HoodieRecordPa
@Override
public HoodieWriteMetadata<JavaRDD<WriteStatus>> performClustering(final HoodieClusteringPlan clusteringPlan, final Schema schema, final String instantTime) {
// execute clustering for each group async and collect WriteStatus
JavaSparkContext engineContext = HoodieSparkEngineContext.getSparkContext(getEngineContext());
// execute clustering for each group async and collect WriteStatus
Stream<JavaRDD<WriteStatus>> writeStatusRDDStream = clusteringPlan.getInputGroups().stream()
Stream<JavaRDD<WriteStatus>> writeStatusRDDStream = FutureUtils.allOf(
clusteringPlan.getInputGroups().stream()
.map(inputGroup -> runClusteringForGroupAsync(inputGroup,
clusteringPlan.getStrategy().getStrategyParams(),
Option.ofNullable(clusteringPlan.getPreserveHoodieMetadata()).orElse(false),
instantTime))
.map(CompletableFuture::join);
.collect(Collectors.toList()))
.join()
.stream();
JavaRDD<WriteStatus>[] writeStatuses = convertStreamToArray(writeStatusRDDStream);
JavaRDD<WriteStatus> writeStatusRDD = engineContext.union(writeStatuses);
@@ -145,7 +147,6 @@ public abstract class MultipleSparkJobExecutionStrategy<T extends HoodieRecordPa
}
}
/**
* Submit job to execute clustering for the group.
*/