Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[compaction]Add local sample config for sort compaction #3081

Merged
merged 5 commits into from
Apr 2, 2024
Merged

[compaction]Add local sample config for sort compaction #3081

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ca63c63
add sample config
wg1026688210 Mar 23, 2024
04d1350
add validation and ut
wg1026688210 Mar 23, 2024
ccea659
add validation and ut
wg1026688210 Mar 23, 2024
5405ab3
refactor
wg1026688210 Mar 25, 2024
032a0e6
refactor
wg1026688210 Mar 27, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions docs/layouts/shortcodes/generated/core_configuration.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -563,13 +563,31 @@
<td>Duration</td>
<td>In watermarking, if a source remains idle beyond the specified timeout duration, it triggers snapshot advancement and facilitates tag creation.</td>
</tr>
<tr>
<td><h5>sort-compaction.global-sample.size</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>Integer</td>
<td>The size of global sample for sort-compaction.By default,the value is the sink parallelism * 1000.</td>
</tr>
<tr>
<td><h5>sort-compaction.local-sample.size</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>Integer</td>
<td>The size of local sample for sort-compaction.By default,the value is the sink parallelism * 1000.</td>
</tr>
<tr>
<td><h5>sort-compaction.range-strategy</h5></td>
<td style="word-wrap: break-word;">QUANTITY</td>
<td><p>Enum</p></td>
<td>The range strategy of sort compaction, the default value is quantity.
If the data size allocated for the sorting task is uneven,which may lead to performance bottlenecks, the config can be set to size.<br /><br />Possible values:<ul><li>"SIZE"</li><li>"QUANTITY"</li></ul></td>
</tr>
<tr>
<td><h5>sort-compaction.range.size</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>Integer</td>
<td>The range for sort-compaction.By default,the value is the sink parallelism * 10.</td>
</tr>
<tr>
<td><h5>sort-engine</h5></td>
<td style="word-wrap: break-word;">loser-tree</td>
Expand Down
33 changes: 33 additions & 0 deletions paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1056,6 +1056,27 @@ 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 =
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

SORT_COMPACTION_SAMPLE_MAGNIFICATION

key("sort-compaction.local-sample.size")
.intType()
.noDefaultValue()
.withDescription(
"The size of local sample for sort-compaction.By default,the value is the sink parallelism * 1000.");
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the parameter could be magnification. int localSampleSize = parallelism * magnification. magnification should be greater than or equal to 20


public static final ConfigOption<Integer> SORT_COMPACTION_GLOBAL_SAMPLE_SIZE =
key("sort-compaction.global-sample.size")
.intType()
.noDefaultValue()
.withDescription(
"The size of global sample for sort-compaction.By default,the value is the sink parallelism * 1000.");

public static final ConfigOption<Integer> SORT_RANGE =
key("sort-compaction.range.size")
.intType()
.noDefaultValue()
.withDescription(
"The range for sort-compaction.By default,the value is the sink parallelism * 10.");

private final Options options;

public CoreOptions(Map<String, String> options) {
Expand Down Expand Up @@ -1126,6 +1147,18 @@ public boolean sortBySize() {
return options.get(SORT_RANG_STRATEGY) == RangeStrategy.SIZE;
}

public Optional<Integer> getLocalSampleSize() {
return options.getOptional(SORT_COMPACTION_LOCAL_SAMPLE_SIZE);
}

public Optional<Integer> getGlobalSampleSize() {
return options.getOptional(SORT_COMPACTION_GLOBAL_SAMPLE_SIZE);
}

public Optional<Integer> getSortRange() {
return options.getOptional(SORT_RANGE);
}

public static FileFormat createFileFormat(
Options options, ConfigOption<FileFormatType> formatOption) {
String formatIdentifier = options.get(formatOption).toString();
Expand Down
7 changes: 4 additions & 3 deletions ...flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/shuffle/RangeShuffle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,8 @@ public static <T> DataStream<Tuple2<T, RowData>> rangeShuffleByKey(
DataStream<Tuple2<T, RowData>> inputDataStream,
SerializableSupplier<Comparator<T>> keyComparator,
TypeInformation<T> keyTypeInformation,
int sampleSize,
int localSampleSize,
int globalSampleSize,
int rangeNum,
int outParallelism,
RowType valueRowType,
Expand All @@ -116,7 +117,7 @@ public static <T> DataStream<Tuple2<T, RowData>> rangeShuffleByKey(
new OneInputTransformation<>(
keyInput,
"LOCAL SAMPLE",
new LocalSampleOperator<>(sampleSize),
new LocalSampleOperator<>(localSampleSize),
new TupleTypeInfo<>(
BasicTypeInfo.DOUBLE_TYPE_INFO,
keyTypeInformation,
Expand All @@ -128,7 +129,7 @@ public static <T> DataStream<Tuple2<T, RowData>> rangeShuffleByKey(
new OneInputTransformation<>(
localSample,
"GLOBAL SAMPLE",
new GlobalSampleOperator<>(sampleSize, keyComparator, rangeNum),
new GlobalSampleOperator<>(globalSampleSize, keyComparator, rangeNum),
new ListTypeInfo<>(keyTypeInformation),
1);

Expand Down
28 changes: 25 additions & 3 deletions paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,8 +99,13 @@ 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 rangeNum = sinkParallelism * 10;
final int localSampleSize =
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

localSampleSize = sinkParallelism * sampleSignification

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done~

options.getLocalSampleSize().orElseGet(() -> sinkParallelism * 1000);
final int globalSampleSize =
options.getGlobalSampleSize().orElseGet(() -> sinkParallelism * 1000);
final int rangeNum = options.getSortRange().orElseGet(() -> sinkParallelism * 10);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.


validateSampleConfig(localSampleSize, globalSampleSize, rangeNum, sinkParallelism);

int keyFieldCount = sortKeyType.getFieldCount();
int valueFieldCount = valueRowType.getFieldCount();
Expand Down Expand Up @@ -144,7 +149,8 @@ public Tuple2<KEY, RowData> map(RowData value) {
inputWithKey,
shuffleKeyComparator,
keyTypeInformation,
sampleSize,
localSampleSize,
globalSampleSize,
rangeNum,
sinkParallelism,
valueRowType,
Expand Down Expand Up @@ -188,6 +194,22 @@ public InternalRow map(InternalRow value) {
.setParallelism(sinkParallelism);
}

private static void validateSampleConfig(
int localSampleSize, int globalSampleSize, int rangeNum, int sinkParallelism) {
if (globalSampleSize < rangeNum) {
throw new IllegalArgumentException(
String.format(
"The global sample size %d should be greater than rangeNum %d.",
globalSampleSize, rangeNum));
}
if (sinkParallelism * localSampleSize < globalSampleSize) {
throw new IllegalArgumentException(
String.format(
"The sum size %d of local sample must be greater than the global sample %d.",
localSampleSize * sinkParallelism, globalSampleSize));
}
}

/** Abstract key from a row data. */
interface KeyAbstract<KEY> extends Serializable {
default void open() {}
Expand Down
93 changes: 78 additions & 15 deletions ...src/test/java/org/apache/paimon/flink/action/SortCompactActionForUnawareBucketITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,8 @@
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.types.DataTypes;

import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;

import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;

Expand Down Expand Up @@ -359,22 +361,83 @@ private void order(List<String> columns) throws Exception {

private SortCompactAction createAction(
String orderStrategy, String rangeStrategy, List<String> columns) {
return createAction(orderStrategy, rangeStrategy, columns, Lists.newArrayList());
}

private SortCompactAction createAction(
String orderStrategy,
String rangeStrategy,
List<String> columns,
List<String> extraConfigs) {
ArrayList<String> args =
Lists.newArrayList(
"compact",
"--warehouse",
warehouse,
"--database",
database,
"--table",
tableName,
"--order_strategy",
orderStrategy,
"--order_by",
String.join(",", columns),
"--table_conf",
"sort-compaction.range-strategy=" + rangeStrategy);
args.addAll(extraConfigs);
return createAction(SortCompactAction.class, args.toArray(new String[0]));
}

@Test
public void testSampleConfig() throws Exception {
prepareData(300, 1);

return createAction(
SortCompactAction.class,
"compact",
"--warehouse",
warehouse,
"--database",
database,
"--table",
tableName,
"--order_strategy",
orderStrategy,
"--order_by",
String.join(",", columns),
"--table_conf sort-compaction.range-strategy=" + rangeStrategy,
rangeStrategy);
{
ArrayList<String> extraCompactionConfig =
Lists.newArrayList(
"--table_conf",
"sort-compaction.range.size=100",
"--table_conf",
"sort-compaction.global-sample.size=10");
Assertions.assertThatCode(
() -> {
createAction(
"order",
"size",
Arrays.asList(
"f0", "f1", "f2", "f3", "f4", "f5", "f6",
"f7", "f8", "f9", "f10", "f11", "f12",
"f13", "f14", "f15"),
extraCompactionConfig)
.run();
})
.hasMessage("The global sample size 10 should be greater than rangeNum 100.");
}

{
ArrayList<String> extraCompactionConfig =
Lists.newArrayList(
"--table_conf",
"sort-compaction.local-sample.size=1",
"--table_conf",
"sort-compaction.global-sample.size=100",
"--table_conf",
"sink.parallelism=3");
Assertions.assertThatCode(
() -> {
createAction(
"order",
"size",
Arrays.asList(
"f0", "f1", "f2", "f3", "f4", "f5", "f6",
"f7", "f8", "f9", "f10", "f11", "f12",
"f13", "f14", "f15"),
extraCompactionConfig)
.run();
})
.hasMessage(
"The sum size 3 of local sample must be greater than the global sample 100.");
}
}

private void callProcedure(
Expand Down
Loading