[compaction]Add local sample config for sort compaction #3081
Conversation
| options.getLocalSampleSize().orElseGet(() -> sinkParallelism * 1000); | ||
| final int globalSampleSize = | ||
| options.getGlobalSampleSize().orElseGet(() -> sinkParallelism * 1000); | ||
| final int rangeNum = options.getSortRange().orElseGet(() -> sinkParallelism * 10); |
There was a problem hiding this comment.
I think there is no need to make localSampleSize and globalSampleSize different, if you want a performance improvement when parallelism is large, just add one parameter of sampleSize option. 20 or 2000 times, default 1000. rangeNum still sinkParallelism * 10.
|
@leaves12138 cc |
| .intType() | ||
| .noDefaultValue() | ||
| .withDescription( | ||
| "The size of local sample for sort-compaction.By default,the value is the sink parallelism * 1000."); |
There was a problem hiding this comment.
I think the parameter could be magnification. int localSampleSize = parallelism * magnification. magnification should be greater than or equal to 20
| @@ -1056,6 +1056,12 @@ public class CoreOptions implements Serializable { | |||
| + "If the data size allocated for the sorting task is uneven,which may lead to performance bottlenecks, " | |||
| + "the config can be set to size."); | |||
|
|
|||
| public static final ConfigOption<Integer> SORT_COMPACTION_LOCAL_SAMPLE_SIZE = | |||
There was a problem hiding this comment.
SORT_COMPACTION_SAMPLE_MAGNIFICATION
| final int rangeNum = sinkParallelism * 10; | ||
|
|
||
| if (localSampleSize < 20) { |
There was a problem hiding this comment.
if(localSampleSignification < 20) throws
| @@ -99,9 +99,18 @@ public static <KEY> DataStream<RowData> sortStreamByKey( | |||
| "The adaptive batch scheduler is not supported. Please set the sink parallelism using the key: " | |||
| + FlinkConnectorOptions.SINK_PARALLELISM.key()); | |||
| } | |||
| final int sampleSize = sinkParallelism * 1000; | |||
| final int localSampleSize = | |||
There was a problem hiding this comment.
localSampleSize = sinkParallelism * sampleSignification
wg1026688210
changed the title
|
@leaves12138 PHAL~ |
wg1026688210
force-pushed
the
sort_compaction/add_sample_config
branch
from
17e242f
to
032a0e6
Compare
Purpose
Add local sample config for sort compaction . It' helpful for the sort compaction task with large parallelism,which sampling may lead to a performance bottleneck.