lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[MINOR] Sync UpsertPartitioner modify of HUDI-1398 to flink/java (#2390)

Browse Source 
Co-authored-by: zhang wen <wen.zhang@dmall.com>
This commit is contained in:
steven zhang
2020-12-31 11:06:41 +08:00
committed by GitHub
parent c5e8a024f6
commit a23aa41a1a
2 changed files with 10 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -181,7 +181,11 @@ public class UpsertPartitioner<T extends HoodieRecordPayload<T>> implements Part
+ ", totalInsertBuckets => " + insertBuckets + ", recordsPerBucket => " + insertRecordsPerBucket); + ", totalInsertBuckets => " + insertBuckets + ", recordsPerBucket => " + insertRecordsPerBucket);
for (int b = 0; b < insertBuckets; b++) { for (int b = 0; b < insertBuckets; b++) {
bucketNumbers.add(totalBuckets); bucketNumbers.add(totalBuckets);
recordsPerBucket.add(totalUnassignedInserts / insertBuckets); if (b < insertBuckets - 1) {
recordsPerBucket.add(insertRecordsPerBucket);
} else {
recordsPerBucket.add(totalUnassignedInserts - (insertBuckets - 1) * insertRecordsPerBucket);
}
BucketInfo bucketInfo = new BucketInfo(); BucketInfo bucketInfo = new BucketInfo();
bucketInfo.bucketType = BucketType.INSERT; bucketInfo.bucketType = BucketType.INSERT;
bucketInfo.partitionPath = partitionPath; bucketInfo.partitionPath = partitionPath;

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

@@ -179,7 +179,11 @@ public class UpsertPartitioner<T extends HoodieRecordPayload<T>> implements Part
+ ", totalInsertBuckets => " + insertBuckets + ", recordsPerBucket => " + insertRecordsPerBucket); + ", totalInsertBuckets => " + insertBuckets + ", recordsPerBucket => " + insertRecordsPerBucket);
for (int b = 0; b < insertBuckets; b++) { for (int b = 0; b < insertBuckets; b++) {
bucketNumbers.add(totalBuckets); bucketNumbers.add(totalBuckets);
recordsPerBucket.add(totalUnassignedInserts / insertBuckets); if (b < insertBuckets - 1) {
recordsPerBucket.add(insertRecordsPerBucket);
} else {
recordsPerBucket.add(totalUnassignedInserts - (insertBuckets - 1) * insertRecordsPerBucket);
}
BucketInfo bucketInfo = new BucketInfo(); BucketInfo bucketInfo = new BucketInfo();
bucketInfo.bucketType = BucketType.INSERT; bucketInfo.bucketType = BucketType.INSERT;
bucketInfo.partitionPath = partitionPath; bucketInfo.partitionPath = partitionPath;