lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-1921] Add target io option for flink compaction (#2980)

Browse Source
This commit is contained in:
Danny Chan
2021-06-02 10:10:35 +08:00
committed by GitHub
parent bf1cfb5635
commit 7fa2f8ea82
2 changed files with 7 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java
View File

@@ -369,6 +369,12 @@ public class FlinkOptions {
.defaultValue(100) // default 100 MB .defaultValue(100) // default 100 MB
.withDescription("Max memory in MB for compaction spillable map, default 100MB"); .withDescription("Max memory in MB for compaction spillable map, default 100MB");
public static final ConfigOption<Long> COMPACTION_TARGET_IO = ConfigOptions
.key("compaction.target_io")
.longType()
.defaultValue(5120L) // default 5 GB
.withDescription("Target IO per compaction (both read and write), default 5 GB");
public static final ConfigOption<Boolean> CLEAN_ASYNC_ENABLED = ConfigOptions public static final ConfigOption<Boolean> CLEAN_ASYNC_ENABLED = ConfigOptions
.key("clean.async.enabled") .key("clean.async.enabled")
.booleanType() .booleanType()

1
hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java
View File

@@ -195,6 +195,7 @@ public class StreamerUtil {
.withCompactionConfig( .withCompactionConfig(
HoodieCompactionConfig.newBuilder() HoodieCompactionConfig.newBuilder()
.withPayloadClass(conf.getString(FlinkOptions.PAYLOAD_CLASS)) .withPayloadClass(conf.getString(FlinkOptions.PAYLOAD_CLASS))
.withTargetIOPerCompactionInMB(conf.getLong(FlinkOptions.COMPACTION_TARGET_IO))
.withInlineCompactionTriggerStrategy( .withInlineCompactionTriggerStrategy(
CompactionTriggerStrategy.valueOf(conf.getString(FlinkOptions.COMPACTION_TRIGGER_STRATEGY).toUpperCase(Locale.ROOT))) CompactionTriggerStrategy.valueOf(conf.getString(FlinkOptions.COMPACTION_TRIGGER_STRATEGY).toUpperCase(Locale.ROOT)))
.withMaxNumDeltaCommitsBeforeCompaction(conf.getInteger(FlinkOptions.COMPACTION_DELTA_COMMITS)) .withMaxNumDeltaCommitsBeforeCompaction(conf.getInteger(FlinkOptions.COMPACTION_DELTA_COMMITS))