lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

Down the reader mem check

Browse Source
This commit is contained in:
v-zhangjc9
2022-06-30 19:13:51 +08:00
parent 215a794fd3
commit 6be03ca56a
1 changed files with 2 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java
View File

@@ -325,10 +325,10 @@ public class StreamWriteFunction<I> extends AbstractStreamWriteFunction<I> {
private final double maxBufferSize; private final double maxBufferSize;
TotalSizeTracer(Configuration conf) { TotalSizeTracer(Configuration conf) {
long mergeReaderMem = 100; // constant 100MB long mergeReaderMem = 10; // constant 100MB
long mergeMapMaxMem = conf.getInteger(FlinkOptions.WRITE_MERGE_MAX_MEMORY); long mergeMapMaxMem = conf.getInteger(FlinkOptions.WRITE_MERGE_MAX_MEMORY);
this.maxBufferSize = (conf.getDouble(FlinkOptions.WRITE_TASK_MAX_SIZE) - mergeReaderMem - mergeMapMaxMem) * 1024 * 1024; this.maxBufferSize = (conf.getDouble(FlinkOptions.WRITE_TASK_MAX_SIZE) - mergeReaderMem - mergeMapMaxMem) * 1024 * 1024;
final String errMsg = String.format("'%s' should be at least greater than '%s' plus merge reader memory(constant 100MB now)", final String errMsg = String.format("'%s' should be at least greater than '%s' plus merge reader memory(constant 50MB now)",
FlinkOptions.WRITE_TASK_MAX_SIZE.key(), FlinkOptions.WRITE_MERGE_MAX_MEMORY.key()); FlinkOptions.WRITE_TASK_MAX_SIZE.key(), FlinkOptions.WRITE_MERGE_MAX_MEMORY.key());
ValidationUtils.checkState(this.maxBufferSize > 0, errMsg); ValidationUtils.checkState(this.maxBufferSize > 0, errMsg);
} }