From 71ce021fc805fac6c8a29ca7f89318ced11b2869 Mon Sep 17 00:00:00 2001 From: Bo Date: Fri, 25 Feb 2022 20:54:42 +0800 Subject: [PATCH 1/3] [HUDI-3460]Flink TM memory Optimization --- .../hudi/configuration/FlinkOptions.java | 14 ++++++++++++++ .../apache/hudi/table/HoodieTableSource.java | 4 +--- .../apache/hudi/table/format/FormatUtils.java | 8 ++++---- .../format/mor/MergeOnReadInputFormat.java | 2 +- .../org/apache/hudi/util/StreamerUtil.java | 19 +++++++++++++++++++ .../hudi/table/TestHoodieTableFactory.java | 2 ++ .../hudi/table/TestHoodieTableSource.java | 3 +++ .../hudi/table/format/TestInputFormat.java | 2 ++ 8 files changed, 46 insertions(+), 8 deletions(-) diff --git a/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java b/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java index 1be90603605cd..dbdeef2559ac1 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java +++ b/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java @@ -18,6 +18,7 @@ package org.apache.hudi.configuration; +import org.apache.flink.configuration.MemorySize; import org.apache.hudi.common.config.ConfigClassProperty; import org.apache.hudi.common.config.ConfigGroups; import org.apache.hudi.common.config.HoodieConfig; @@ -26,6 +27,7 @@ import org.apache.hudi.config.HoodieIndexConfig; import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.hadoop.config.HoodieRealtimeConfig; import org.apache.hudi.hive.SlashEncodedDayPartitionValueExtractor; import org.apache.hudi.index.HoodieIndex; import org.apache.hudi.keygen.constant.KeyGeneratorOptions; @@ -481,6 +483,18 @@ private FlinkOptions() { // Compaction Options // ------------------------------------------------------------------------ + public static final ConfigOption COMPACTION_MEMORY_FRACTION_PROP = ConfigOptions + .key(HoodieRealtimeConfig.COMPACTION_MEMORY_FRACTION_PROP) + .doubleType() + .defaultValue(0.1) + .withDescription("Fraction of Flink TM memory used for compaction of log files."); + + public static final ConfigOption COMPACTION_MEMORY_SIZE = ConfigOptions + .key("compaction.memory.size") + .memoryType() + .noDefaultValue() + .withDescription("Specify the memory of Flink TM used for compaction of log files."); + public static final ConfigOption COMPACTION_SCHEDULE_ENABLED = ConfigOptions .key("compaction.schedule.enabled") .booleanType() diff --git a/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java b/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java index 3efd1d5612f15..d00eb3e3ec700 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java @@ -75,7 +75,6 @@ import org.apache.flink.table.types.logical.RowType; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.Path; -import org.apache.hadoop.mapred.JobConf; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -93,7 +92,6 @@ import java.util.stream.Collectors; import java.util.stream.IntStream; -import static org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes; import static org.apache.hudi.table.format.FormatUtils.getParquetConf; /** @@ -159,7 +157,7 @@ public HoodieTableSource( this.filters = filters == null ? Collections.emptyList() : filters; this.hadoopConf = StreamerUtil.getHadoopConf(); this.metaClient = StreamerUtil.metaClientForReader(conf, hadoopConf); - this.maxCompactionMemoryInBytes = getMaxCompactionMemoryInBytes(new JobConf(this.hadoopConf)); + this.maxCompactionMemoryInBytes = StreamerUtil.getMaxCompactionMemoryInBytes(conf); } @Override diff --git a/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java b/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java index fa404cc2163ec..666dc3a73fb90 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java @@ -32,7 +32,6 @@ import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.hadoop.config.HoodieRealtimeConfig; -import org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils; import org.apache.hudi.table.format.mor.MergeOnReadInputSplit; import org.apache.avro.Schema; @@ -43,7 +42,7 @@ import org.apache.flink.types.RowKind; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.mapred.JobConf; +import org.apache.hudi.util.StreamerUtil; import java.util.ArrayList; import java.util.Arrays; @@ -190,9 +189,10 @@ public static class BoundedMemoryRecords { public BoundedMemoryRecords( MergeOnReadInputSplit split, Schema logSchema, - Configuration hadoopConf) { + Configuration hadoopConf, + org.apache.flink.configuration.Configuration flinkConf) { this.executor = new BoundedInMemoryExecutor<>( - HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes(new JobConf(hadoopConf)), + StreamerUtil.getMaxCompactionMemoryInBytes(flinkConf), getParallelProducers(), Option.empty(), Function.identity(), diff --git a/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputFormat.java b/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputFormat.java index 4404e15eaaccf..4ea202a70b986 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputFormat.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputFormat.java @@ -403,7 +403,7 @@ private ClosableIterator getUnMergedLogFileIterator(MergeOnReadInputSpl final GenericRecordBuilder recordBuilder = new GenericRecordBuilder(requiredSchema); final AvroToRowDataConverters.AvroToRowDataConverter avroToRowDataConverter = AvroToRowDataConverters.createRowConverter(tableState.getRequiredRowType()); - final FormatUtils.BoundedMemoryRecords records = new FormatUtils.BoundedMemoryRecords(split, tableSchema, hadoopConf); + final FormatUtils.BoundedMemoryRecords records = new FormatUtils.BoundedMemoryRecords(split, tableSchema, hadoopConf, conf); final Iterator> recordsIterator = records.getRecordsIterator(); return new ClosableIterator() { diff --git a/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java b/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java index 45d23f2ff4ea1..be5bb86ef67d3 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java +++ b/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java @@ -18,6 +18,8 @@ package org.apache.hudi.util; +import org.apache.flink.configuration.TaskManagerOptions; +import org.apache.flink.runtime.clusterframework.TaskExecutorProcessUtils; import org.apache.hudi.client.FlinkTaskContextSupplier; import org.apache.hudi.client.HoodieFlinkWriteClient; import org.apache.hudi.client.common.HoodieFlinkEngineContext; @@ -81,6 +83,8 @@ import static org.apache.hudi.common.model.HoodieFileFormat.ORC; import static org.apache.hudi.common.model.HoodieFileFormat.PARQUET; import static org.apache.hudi.common.table.HoodieTableConfig.ARCHIVELOG_FOLDER; +import static org.apache.hudi.configuration.FlinkOptions.COMPACTION_MEMORY_FRACTION_PROP; +import static org.apache.hudi.configuration.FlinkOptions.COMPACTION_MEMORY_SIZE; /** * Utilities for Flink stream read and write. @@ -509,4 +513,19 @@ public static String getLastCompletedInstant(HoodieTableMetaClient metaClient) { public static boolean haveSuccessfulCommits(HoodieTableMetaClient metaClient) { return !metaClient.getCommitsTimeline().filterCompletedInstants().empty(); } + + /** + * get the max compaction memory in bytes from conf. + */ + public static long getMaxCompactionMemoryInBytes(Configuration conf) { + if (conf.contains(COMPACTION_MEMORY_SIZE)) { + return conf.get(COMPACTION_MEMORY_SIZE).getBytes(); + } + return (long)Math + .ceil(conf.getDouble(COMPACTION_MEMORY_FRACTION_PROP) + * TaskExecutorProcessUtils.processSpecFromConfig( + TaskExecutorProcessUtils.getConfigurationMapLegacyTaskManagerHeapSizeToConfigOption( + conf, TaskManagerOptions.TOTAL_PROCESS_MEMORY)) + .getManagedMemorySize().getBytes()); + } } diff --git a/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java b/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java index a76e00816189a..78a95eb765616 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java +++ b/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java @@ -18,6 +18,7 @@ package org.apache.hudi.table; +import org.apache.flink.configuration.MemorySize; import org.apache.hudi.common.model.DefaultHoodieRecordPayload; import org.apache.hudi.common.model.EventTimeAvroPayload; import org.apache.hudi.configuration.FlinkOptions; @@ -83,6 +84,7 @@ void beforeEach() throws IOException { this.conf = new Configuration(); this.conf.setString(FlinkOptions.PATH, tempFile.getAbsolutePath()); this.conf.setString(FlinkOptions.TABLE_NAME, "t1"); + this.conf.set(FlinkOptions.COMPACTION_MEMORY_SIZE, MemorySize.ofMebiBytes(1024)); StreamerUtil.initTableIfNotExists(this.conf); } diff --git a/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java b/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java index 8ee18a9601b2f..98dfd94db275f 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java +++ b/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java @@ -18,6 +18,7 @@ package org.apache.hudi.table; +import org.apache.flink.configuration.MemorySize; import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.table.format.mor.MergeOnReadInputFormat; import org.apache.hudi.utils.TestConfigurations; @@ -64,6 +65,7 @@ public class TestHoodieTableSource { void beforeEach() throws Exception { final String path = tempFile.getAbsolutePath(); conf = TestConfigurations.getDefaultConf(path); + conf.set(FlinkOptions.COMPACTION_MEMORY_SIZE, MemorySize.ofMebiBytes(1024)); TestData.writeData(TestData.DATA_SET_INSERT, conf); } @@ -122,6 +124,7 @@ void testGetTableAvroSchema() { final String path = tempFile.getAbsolutePath(); conf = TestConfigurations.getDefaultConf(path); conf.setBoolean(FlinkOptions.READ_AS_STREAMING, true); + conf.set(FlinkOptions.COMPACTION_MEMORY_SIZE, MemorySize.ofMebiBytes(1024)); HoodieTableSource tableSource = new HoodieTableSource( TestConfigurations.TABLE_SCHEMA, diff --git a/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java b/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java index 6fbbab81fa4a6..6390ab9997c83 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java +++ b/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java @@ -18,6 +18,7 @@ package org.apache.hudi.table.format; +import org.apache.flink.configuration.MemorySize; import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.table.HoodieTableMetaClient; import org.apache.hudi.common.table.timeline.HoodieInstant; @@ -73,6 +74,7 @@ void beforeEach(HoodieTableType tableType, Map options) throws I conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); conf.setString(FlinkOptions.TABLE_TYPE, tableType.name()); conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false); // close the async compaction + conf.set(FlinkOptions.COMPACTION_MEMORY_SIZE, MemorySize.ofMebiBytes(1024)); options.forEach((key, value) -> conf.setString(key, value)); StreamerUtil.initTableIfNotExists(conf); From b39990b6ac17dc5db3f4e951cd8f566208abb49f Mon Sep 17 00:00:00 2001 From: Bo Date: Thu, 3 Mar 2022 17:26:50 +0800 Subject: [PATCH 2/3] [HUDI-3460]Flink TM memory Optimization --- .../java/org/apache/hudi/configuration/FlinkOptions.java | 9 +-------- .../apache/hudi/sink/compact/FlinkCompactionConfig.java | 5 +++++ .../org/apache/hudi/streamer/FlinkStreamerConfig.java | 5 +++++ .../src/main/java/org/apache/hudi/util/StreamerUtil.java | 8 ++++---- .../org/apache/hudi/table/TestHoodieTableFactory.java | 3 +-- .../org/apache/hudi/table/TestHoodieTableSource.java | 5 ++--- .../org/apache/hudi/table/format/TestInputFormat.java | 3 +-- 7 files changed, 19 insertions(+), 19 deletions(-) diff --git a/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java b/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java index dbdeef2559ac1..d8b0dc0ce1899 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java +++ b/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java @@ -18,7 +18,6 @@ package org.apache.hudi.configuration; -import org.apache.flink.configuration.MemorySize; import org.apache.hudi.common.config.ConfigClassProperty; import org.apache.hudi.common.config.ConfigGroups; import org.apache.hudi.common.config.HoodieConfig; @@ -487,13 +486,7 @@ private FlinkOptions() { .key(HoodieRealtimeConfig.COMPACTION_MEMORY_FRACTION_PROP) .doubleType() .defaultValue(0.1) - .withDescription("Fraction of Flink TM memory used for compaction of log files."); - - public static final ConfigOption COMPACTION_MEMORY_SIZE = ConfigOptions - .key("compaction.memory.size") - .memoryType() - .noDefaultValue() - .withDescription("Specify the memory of Flink TM used for compaction of log files."); + .withDescription("Compaction memory fraction of Task Manager managed memory size, default 0.1."); public static final ConfigOption COMPACTION_SCHEDULE_ENABLED = ConfigOptions .key("compaction.schedule.enabled") diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/compact/FlinkCompactionConfig.java b/hudi-flink/src/main/java/org/apache/hudi/sink/compact/FlinkCompactionConfig.java index 4f3faadb92f01..41d3179988b52 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/sink/compact/FlinkCompactionConfig.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/compact/FlinkCompactionConfig.java @@ -109,6 +109,10 @@ public class FlinkCompactionConfig extends Configuration { description = "Min compaction interval of async compaction service, default 10 minutes") public Integer minCompactionIntervalSeconds = 600; + @Parameter(names = {"--compaction-memory-fraction-prop"}, + description = "Compaction memory fraction of Task Manager managed memory size, default 0.1") + public double compactionMemoryFractionProp = 0.1; + /** * Transforms a {@code HoodieFlinkCompaction.config} into {@code Configuration}. * The latter is more suitable for the table APIs. It reads all the properties @@ -129,6 +133,7 @@ public static org.apache.flink.configuration.Configuration toFlinkConfig(FlinkCo conf.setLong(FlinkOptions.COMPACTION_TARGET_IO, config.compactionTargetIo); conf.setInteger(FlinkOptions.COMPACTION_TASKS, config.compactionTasks); conf.setBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED, config.cleanAsyncEnable); + conf.setDouble(FlinkOptions.COMPACTION_MEMORY_FRACTION_PROP, config.compactionMemoryFractionProp); // use synchronous compaction always conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false); conf.setBoolean(FlinkOptions.COMPACTION_SCHEDULE_ENABLED, config.schedule); diff --git a/hudi-flink/src/main/java/org/apache/hudi/streamer/FlinkStreamerConfig.java b/hudi-flink/src/main/java/org/apache/hudi/streamer/FlinkStreamerConfig.java index 1d7111f495c58..0892fc5c4caa2 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/streamer/FlinkStreamerConfig.java +++ b/hudi-flink/src/main/java/org/apache/hudi/streamer/FlinkStreamerConfig.java @@ -250,6 +250,10 @@ public class FlinkStreamerConfig extends Configuration { @Parameter(names = {"--compaction-target-io"}, description = "Target IO per compaction (both read and write), default 500 GB") public Long compactionTargetIo = 512000L; + @Parameter(names = {"--compaction-memory-fraction-prop"}, + description = "Compaction memory fraction of Task Manager managed memory size, default 0.1") + public double compactionMemoryFractionProp = 0.1; + @Parameter(names = {"--clean-async-enabled"}, description = "Whether to cleanup the old commits immediately on new commits, enabled by default") public Boolean cleanAsyncEnabled = true; @@ -384,6 +388,7 @@ public static org.apache.flink.configuration.Configuration toFlinkConfig(FlinkSt conf.setInteger(FlinkOptions.COMPACTION_DELTA_SECONDS, config.compactionDeltaSeconds); conf.setInteger(FlinkOptions.COMPACTION_MAX_MEMORY, config.compactionMaxMemory); conf.setLong(FlinkOptions.COMPACTION_TARGET_IO, config.compactionTargetIo); + conf.setDouble(FlinkOptions.COMPACTION_MEMORY_FRACTION_PROP, config.compactionMemoryFractionProp); conf.setBoolean(FlinkOptions.CLEAN_ASYNC_ENABLED, config.cleanAsyncEnabled); conf.setInteger(FlinkOptions.CLEAN_RETAIN_COMMITS, config.cleanRetainCommits); conf.setInteger(FlinkOptions.ARCHIVE_MAX_COMMITS, config.archiveMaxCommits); diff --git a/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java b/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java index be5bb86ef67d3..d661572c93995 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java +++ b/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java @@ -84,7 +84,7 @@ import static org.apache.hudi.common.model.HoodieFileFormat.PARQUET; import static org.apache.hudi.common.table.HoodieTableConfig.ARCHIVELOG_FOLDER; import static org.apache.hudi.configuration.FlinkOptions.COMPACTION_MEMORY_FRACTION_PROP; -import static org.apache.hudi.configuration.FlinkOptions.COMPACTION_MEMORY_SIZE; +import static org.apache.hudi.configuration.FlinkOptions.COMPACTION_MAX_MEMORY; /** * Utilities for Flink stream read and write. @@ -515,11 +515,11 @@ public static boolean haveSuccessfulCommits(HoodieTableMetaClient metaClient) { } /** - * get the max compaction memory in bytes from conf. + * Returns the max compaction memory in bytes with given conf. */ public static long getMaxCompactionMemoryInBytes(Configuration conf) { - if (conf.contains(COMPACTION_MEMORY_SIZE)) { - return conf.get(COMPACTION_MEMORY_SIZE).getBytes(); + if (conf.contains(COMPACTION_MAX_MEMORY)) { + return conf.get(COMPACTION_MAX_MEMORY) * 1024 * 1024; } return (long)Math .ceil(conf.getDouble(COMPACTION_MEMORY_FRACTION_PROP) diff --git a/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java b/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java index 78a95eb765616..b3b76d0d7ba6b 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java +++ b/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java @@ -18,7 +18,6 @@ package org.apache.hudi.table; -import org.apache.flink.configuration.MemorySize; import org.apache.hudi.common.model.DefaultHoodieRecordPayload; import org.apache.hudi.common.model.EventTimeAvroPayload; import org.apache.hudi.configuration.FlinkOptions; @@ -84,7 +83,7 @@ void beforeEach() throws IOException { this.conf = new Configuration(); this.conf.setString(FlinkOptions.PATH, tempFile.getAbsolutePath()); this.conf.setString(FlinkOptions.TABLE_NAME, "t1"); - this.conf.set(FlinkOptions.COMPACTION_MEMORY_SIZE, MemorySize.ofMebiBytes(1024)); + this.conf.set(FlinkOptions.COMPACTION_MAX_MEMORY, 1024); StreamerUtil.initTableIfNotExists(this.conf); } diff --git a/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java b/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java index 98dfd94db275f..c41040a56c115 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java +++ b/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java @@ -18,7 +18,6 @@ package org.apache.hudi.table; -import org.apache.flink.configuration.MemorySize; import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.table.format.mor.MergeOnReadInputFormat; import org.apache.hudi.utils.TestConfigurations; @@ -65,7 +64,7 @@ public class TestHoodieTableSource { void beforeEach() throws Exception { final String path = tempFile.getAbsolutePath(); conf = TestConfigurations.getDefaultConf(path); - conf.set(FlinkOptions.COMPACTION_MEMORY_SIZE, MemorySize.ofMebiBytes(1024)); + conf.set(FlinkOptions.COMPACTION_MAX_MEMORY, 1024); TestData.writeData(TestData.DATA_SET_INSERT, conf); } @@ -124,7 +123,7 @@ void testGetTableAvroSchema() { final String path = tempFile.getAbsolutePath(); conf = TestConfigurations.getDefaultConf(path); conf.setBoolean(FlinkOptions.READ_AS_STREAMING, true); - conf.set(FlinkOptions.COMPACTION_MEMORY_SIZE, MemorySize.ofMebiBytes(1024)); + conf.set(FlinkOptions.COMPACTION_MAX_MEMORY, 1024); HoodieTableSource tableSource = new HoodieTableSource( TestConfigurations.TABLE_SCHEMA, diff --git a/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java b/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java index 6390ab9997c83..1e6029a2dd638 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java +++ b/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java @@ -18,7 +18,6 @@ package org.apache.hudi.table.format; -import org.apache.flink.configuration.MemorySize; import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.table.HoodieTableMetaClient; import org.apache.hudi.common.table.timeline.HoodieInstant; @@ -74,7 +73,7 @@ void beforeEach(HoodieTableType tableType, Map options) throws I conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); conf.setString(FlinkOptions.TABLE_TYPE, tableType.name()); conf.setBoolean(FlinkOptions.COMPACTION_ASYNC_ENABLED, false); // close the async compaction - conf.set(FlinkOptions.COMPACTION_MEMORY_SIZE, MemorySize.ofMebiBytes(1024)); + conf.set(FlinkOptions.COMPACTION_MAX_MEMORY, 1024); options.forEach((key, value) -> conf.setString(key, value)); StreamerUtil.initTableIfNotExists(conf); From 0f152f27ca04417f9c2ef159da0c55f5f013fead Mon Sep 17 00:00:00 2001 From: Bo Date: Fri, 4 Mar 2022 14:38:28 +0800 Subject: [PATCH 3/3] [HUDI-3460]Flink TM memory Optimization --- .../src/main/java/org/apache/hudi/util/StreamerUtil.java | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java b/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java index d661572c93995..8861b43882c2c 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java +++ b/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java @@ -83,8 +83,6 @@ import static org.apache.hudi.common.model.HoodieFileFormat.ORC; import static org.apache.hudi.common.model.HoodieFileFormat.PARQUET; import static org.apache.hudi.common.table.HoodieTableConfig.ARCHIVELOG_FOLDER; -import static org.apache.hudi.configuration.FlinkOptions.COMPACTION_MEMORY_FRACTION_PROP; -import static org.apache.hudi.configuration.FlinkOptions.COMPACTION_MAX_MEMORY; /** * Utilities for Flink stream read and write. @@ -518,11 +516,11 @@ public static boolean haveSuccessfulCommits(HoodieTableMetaClient metaClient) { * Returns the max compaction memory in bytes with given conf. */ public static long getMaxCompactionMemoryInBytes(Configuration conf) { - if (conf.contains(COMPACTION_MAX_MEMORY)) { - return conf.get(COMPACTION_MAX_MEMORY) * 1024 * 1024; + if (conf.contains(FlinkOptions.COMPACTION_MAX_MEMORY)) { + return conf.get(FlinkOptions.COMPACTION_MAX_MEMORY) * 1024 * 1024; } return (long)Math - .ceil(conf.getDouble(COMPACTION_MEMORY_FRACTION_PROP) + .ceil(conf.getDouble(FlinkOptions.COMPACTION_MEMORY_FRACTION_PROP) * TaskExecutorProcessUtils.processSpecFromConfig( TaskExecutorProcessUtils.getConfigurationMapLegacyTaskManagerHeapSizeToConfigOption( conf, TaskManagerOptions.TOTAL_PROCESS_MEMORY))