lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-2106] Fix flink batch compaction bug while user don't set compaction tasks (#3192)

Browse Source
This commit is contained in:
swuferhong
2021-07-06 09:10:37 +08:00
committed by GitHub
parent 32bd8ce088
commit bc313727e3
2 changed files with 4 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
hudi-flink/src/main/java/org/apache/hudi/sink/compact/FlinkCompactionConfig.java
View File

@@ -86,8 +86,8 @@ public class FlinkCompactionConfig extends Configuration {
@Parameter(names = {"--compaction-target-io"}, description = "Target IO per compaction (both read and write) for batching compaction, default 512000M.", required = false) @Parameter(names = {"--compaction-target-io"}, description = "Target IO per compaction (both read and write) for batching compaction, default 512000M.", required = false)
public Long compactionTargetIo = 512000L; public Long compactionTargetIo = 512000L;
@Parameter(names = {"--compaction-tasks"}, description = "Parallelism of tasks that do actual compaction, default is 10", required = false) @Parameter(names = {"--compaction-tasks"}, description = "Parallelism of tasks that do actual compaction, default is -1", required = false)
public Integer compactionTasks = 10; public Integer compactionTasks = -1;
/** /**
* Transforms a {@code HoodieFlinkCompaction.config} into {@code Configuration}. * Transforms a {@code HoodieFlinkCompaction.config} into {@code Configuration}.

3
hudi-flink/src/main/java/org/apache/hudi/sink/compact/HoodieFlinkCompactor.java
View File

@@ -112,7 +112,8 @@ public class HoodieFlinkCompactor {
} }
// get compactionParallelism. // get compactionParallelism.
int compactionParallelism = Math.min(conf.getInteger(FlinkOptions.COMPACTION_TASKS), compactionPlan.getOperations().size()); int compactionParallelism = conf.getInteger(FlinkOptions.COMPACTION_TASKS) == -1
? compactionPlan.getOperations().size() : conf.getInteger(FlinkOptions.COMPACTION_TASKS);
env.addSource(new CompactionPlanSourceFunction(table, instant, compactionPlan, compactionInstantTime)) env.addSource(new CompactionPlanSourceFunction(table, instant, compactionPlan, compactionInstantTime))
.name("compaction_source") .name("compaction_source")