lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-2122] Improvement in packaging insert into smallfiles (#3213)

Browse Source
This commit is contained in:
wangxianghu
2021-07-06 00:30:57 +08:00
committed by GitHub
parent 287d2dd79c
commit 650c4455c6
1 changed files with 5 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java
View File

@@ -187,7 +187,7 @@ public class UpsertPartitioner<T extends HoodieRecordPayload<T>> extends Partiti
for (SmallFile smallFile : smallFiles) { for (SmallFile smallFile : smallFiles) {
long recordsToAppend = Math.min((config.getParquetMaxFileSize() - smallFile.sizeBytes) / averageRecordSize, long recordsToAppend = Math.min((config.getParquetMaxFileSize() - smallFile.sizeBytes) / averageRecordSize,
totalUnassignedInserts); totalUnassignedInserts);
if (recordsToAppend > 0 && totalUnassignedInserts > 0) { if (recordsToAppend > 0) {
// create a new bucket or re-use an existing bucket // create a new bucket or re-use an existing bucket
int bucket; int bucket;
if (updateLocationToBucket.containsKey(smallFile.location.getFileId())) { if (updateLocationToBucket.containsKey(smallFile.location.getFileId())) {
@@ -200,6 +200,10 @@ public class UpsertPartitioner<T extends HoodieRecordPayload<T>> extends Partiti
bucketNumbers.add(bucket); bucketNumbers.add(bucket);
recordsPerBucket.add(recordsToAppend); recordsPerBucket.add(recordsToAppend);
totalUnassignedInserts -= recordsToAppend; totalUnassignedInserts -= recordsToAppend;
if (totalUnassignedInserts <= 0) {
// stop the loop when all the inserts are assigned
break;
}
} }
} }