zhangyue19921010 commented on a change in pull request #4753:
URL: https://github.com/apache/hudi/pull/4753#discussion_r800325764
##########
File path:
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDWriteClient.java
##########
@@ -299,7 +299,8 @@ protected void completeCompaction(HoodieCommitMetadata
metadata, JavaRDD<WriteSt
HoodieTable<T, JavaRDD<HoodieRecord<T>>,
JavaRDD<HoodieKey>, JavaRDD<WriteStatus>> table,
String compactionCommitTime) {
this.context.setJobStatus(this.getClass().getSimpleName(), "Collect
compaction write status and commit compaction");
- List<HoodieWriteStat> writeStats =
writeStatuses.map(WriteStatus::getStat).collect();
+ List<HoodieWriteStat> writeStats =
metadata.getPartitionToWriteStats().entrySet().stream().flatMap(e ->
+ e.getValue().stream()).collect(Collectors.toList());
Review comment:
Hope I understand the question correctly, and try to explain the diff
here.
writeStatuses is defined as JavaRDD<WriteStatus>. And
`writeStatuses.map(WriteStatus::getStat).collect()` are RDD computation which
are Lazy. Also writeStatuses is persisted before.
When memory cache is invalid,
`writeStatuses.map(WriteStatus::getStat).collect()` will do a fully action
computation also create new data files with new task ID and Stage ID.
Here use `metadata.getPartitionToWriteStats().xxx` instead of RDD
computation to avoid this problem.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
