MartijnVisser commented on code in PR #126:
URL: https://github.com/apache/flink-benchmarks/pull/126#discussion_r4392936059
##########
src/main/java/org/apache/flink/benchmark/SinkUpsertMaterializerBenchmark.java:
##########
@@ -48,172 +53,84 @@
import org.openjdk.jmh.annotations.Setup;
import org.openjdk.jmh.annotations.State;
import org.openjdk.jmh.annotations.TearDown;
-import org.openjdk.jmh.runner.Runner;
-import org.openjdk.jmh.runner.RunnerException;
-import org.openjdk.jmh.runner.options.OptionsBuilder;
-import org.openjdk.jmh.runner.options.VerboseMode;
-import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.Random;
import static
org.apache.flink.table.runtime.util.StreamRecordUtils.deleteRecord;
import static
org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord;
-import static org.apache.flink.util.Preconditions.checkState;
/**
- * Benchmark for {@link SinkUpsertMaterializer}.
+ * Benchmark for {@link SinkUpsertMaterializer} and {@link
SinkUpsertMaterializerV2}.
*
- * <p>It inserts a fixed number of records (inserts and deletes) in a loop and
measures the time it
- * takes to handle all the records. It uses a test harness, not the full Flink
job.
+ * <p>It uses a test harness, not the full Flink job. All records share the
same stream key. The
+ * setup fills the state of that key with {@code historyLength} rows; each
invocation then inserts
+ * one row and retracts the oldest one, so the history length stays constant
during the measurement.
*
* <p>The benchmark has the following parameters:
*
* <ul>
- * <li>total number of records - fixed to 10K; all records are inserted
under the same stream key
+ * <li>sumVersion: V1/V2 (V2 with the default state strategy)
* <li>hasUpsertKey: true/false
- * <li>stateBackend type: HEAP/RocksDB
- * <li>payload size - fixed to 100 bytes; bigger payload degrades faster on
RocksDB
- * <li>retractPercentage, retractDelay - how many records to retract, and
from which record to
- * start retracting; both parameters control how frequently retraction
happens and how long
- * the history is. Retraction is performed from the middle of the list.
Retraction is
- * inefficient in the current implementation on long lists.
+ * <li>stateBackend: HEAP/ROCKSDB
+ * <li>historyLength: number of rows kept for the stream key; longer
histories make retractions
+ * more expensive
+ * <li>payloadSize: fixed to 20; a bigger payload amplifies inefficiencies,
mostly on RocksDB
* </ul>
- *
- * <p>Results on M3Pro:
- *
- * <pre>{@code
- * Benchmark (hasUpsertKey) (payloadSize)
(retractDelay) (retractPercentage) (stateBackend) (sumVersion) Mode Cnt
Score Error Units
- * SinkUpsertMaterializerBenchmark.run false 100
1 100 HEAP V1 thrpt 4138.808
ops/ms
- * SinkUpsertMaterializerBenchmark.run false 100
1 100 ROCKSDB V1 thrpt 284.055
ops/ms
- * SinkUpsertMaterializerBenchmark.run false 100
10 100 HEAP V1 thrpt 3729.824
ops/ms
- * SinkUpsertMaterializerBenchmark.run false 100
10 100 ROCKSDB V1 thrpt 205.047
ops/ms
- * SinkUpsertMaterializerBenchmark.run false 100
100 100 HEAP V1 thrpt 4137.591
ops/ms
- * SinkUpsertMaterializerBenchmark.run false 100
100 100 ROCKSDB V1 thrpt 80.406
ops/ms
- * SinkUpsertMaterializerBenchmark.run false 100
200 100 HEAP V1 thrpt 1886.574
ops/ms
- * SinkUpsertMaterializerBenchmark.run false 100
200 100 ROCKSDB V1 thrpt 30.935
ops/ms
- * SinkUpsertMaterializerBenchmark.run false 100
1000 100 HEAP V1 thrpt 546.826
ops/ms
- * SinkUpsertMaterializerBenchmark.run false 100
1000 100 ROCKSDB V1 thrpt 7.081
ops/ms
- * SinkUpsertMaterializerBenchmark.run true 100
1 100 HEAP V1 thrpt 4006.263
ops/ms
- * SinkUpsertMaterializerBenchmark.run true 100
1 100 ROCKSDB V1 thrpt 297.556
ops/ms
- * SinkUpsertMaterializerBenchmark.run true 100
10 100 HEAP V1 thrpt 3240.089
ops/ms
- * SinkUpsertMaterializerBenchmark.run true 100
10 100 ROCKSDB V1 thrpt 209.375
ops/ms
- * SinkUpsertMaterializerBenchmark.run true 100
100 100 HEAP V1 thrpt 2131.445
ops/ms
- * SinkUpsertMaterializerBenchmark.run true 100
100 100 ROCKSDB V1 thrpt 78.209
ops/ms
- * SinkUpsertMaterializerBenchmark.run true 100
200 100 HEAP V1 thrpt 652.936
ops/ms
- * SinkUpsertMaterializerBenchmark.run true 100
200 100 ROCKSDB V1 thrpt 29.674
ops/ms
- * SinkUpsertMaterializerBenchmark.run true 100
1000 100 HEAP V1 thrpt 118.567
ops/ms
- * SinkUpsertMaterializerBenchmark.run true 100
1000 100 ROCKSDB V1 thrpt 6.426
ops/ms
- * }</pre>
*/
-@OperationsPerInvocation(value =
SinkUpsertMaterializerBenchmark.RECORDS_PER_INVOCATION)
-@SuppressWarnings("ConstantValue")
public class SinkUpsertMaterializerBenchmark extends BenchmarkBase {
private static final int STREAM_KEY = 0;
- public static void main(String[] args) throws RunnerException {
- new Runner(
- new OptionsBuilder()
- .verbosity(VerboseMode.NORMAL)
- .include(
- ".*"
- +
SinkUpsertMaterializerBenchmark.class
- .getCanonicalName()
- + ".*")
- .build())
- .run();
- }
-
@Benchmark
- public void run(SumBmState state) throws Exception {
- for (long record = 0; record < state.numRecordsTotal; record++) {
- state.harness.processElement(insertRecord(record, STREAM_KEY,
state.payload));
- if (state.shouldRetract(record)) {
- state.harness.processElement(
- deleteRecord(record - state.retractOffset, STREAM_KEY,
state.payload));
- }
- }
+ @OperationsPerInvocation(2)
+ public void sinkUpsertMaterializer(SumBmState state) throws Exception {
+ long record = state.nextRecord++;
+ state.harness.processElement(insertRecord(record, STREAM_KEY,
state.payload));
+ state.harness.processElement(
+ deleteRecord(record - state.historyLength, STREAM_KEY,
state.payload));
}
- protected static final int RECORDS_PER_INVOCATION = 10_000;
-
@State(Scope.Thread)
public static class SumBmState {
+ @Param({"V1", "V2"})
+ public SumVersion sumVersion;
+
@Param({"false", "true"})
public boolean hasUpsertKey;
@Param({"HEAP", "ROCKSDB"})
public SumStateBackend stateBackend;
- public int numRecordsTotal;
+ @Param({"1", "500"})
+ public int historyLength;
- // larger payload amplifies any inefficiencies but slows down the
benchmark; mostly affects
- // rocksdb
@Param({"20"})
public int payloadSize;
- // lower retraction percentage implies longer history, making
retractions even harder
- // (unless percentage = 0)
- @Param("100")
- public int retractPercentage;
-
- // higher retraction delay leaves longer history, making retractions
even harder (unless
- // percentage = 0)
- // for automated runs, reduce the run time (and the data points) to
the most common cases
- @Param({"1", "1000"})
- // for comparison, the following values might be useful:
- // @Param({"1", "10", "100", "200", "1000"})
- public int retractDelay;
-
- // the lower the value, the closer to the end of the list is the
element to retract, the
- // harder for V1 to find the element
- public long retractOffset;
-
public String payload;
- public KeyedOneInputStreamOperatorTestHarness<RowData, RowData,
RowData> harness;
-
- @Setup(Level.Invocation)
- public void initSumBmState() throws Exception {
- harness = getHarness(createSum(hasUpsertKey), stateBackend);
- payload = generatePayload(payloadSize);
- numRecordsTotal = RECORDS_PER_INVOCATION;
- retractOffset = (1 + retractDelay) / 2;
Review Comment:
It now retracts the middle row. #106 hit the oldest row too, since
`retractPercentage` 100 skips `retractDelay`. LEGACY on HEAP without an upsert
key at 500 rows drops from 9850 to 2435 ops/ms.
##########
src/main/java/org/apache/flink/benchmark/SinkUpsertMaterializerBenchmark.java:
##########
@@ -230,21 +147,46 @@ private static
KeyedOneInputStreamOperatorTestHarness<RowData, RowData, RowData>
materializer, rowDataSelector,
rowDataSelector.getProducedType());
testHarness.getExecutionConfig().setMaxParallelism(2048);
testHarness.setStateBackend(stateBackend.create(true));
- testHarness.setOutputCreator(ign -> new BlackHoleOutput()); //
requires change in Flink
+ testHarness.setOutputCreator(ign -> new BlackHoleOutput());
testHarness.open();
return testHarness;
}
- private static OneInputStreamOperator<RowData, RowData> createSum(boolean
hasUpsertKey) {
- StateTtlConfig ttlConfig = StateConfigUtil.createTtlConfig(0); // no
ttl
- RowType physicalRowType = RowType.of(types);
- int[] inputUpsertKey = hasUpsertKey ? new int[] {UPSERT_KEY_POS} :
null;
- return new SinkUpsertMaterializer(
- ttlConfig,
- InternalSerializers.create(physicalRowType),
- equalizer(),
- hasUpsertKey ? upsertEqualizer() : null,
- inputUpsertKey);
+ public enum SumVersion {
+ V1 {
+ @Override
+ OneInputStreamOperator<RowData, RowData> create(boolean
hasUpsertKey) {
+ return SinkUpsertMaterializer.create(
+ TTL_CONFIG,
+ RowType.of(types),
+ equalizer(),
+ hasUpsertKey ? upsertEqualizer() : null,
+ upsertKey(hasUpsertKey));
+ }
+ },
+ V2 {
+ @Override
+ OneInputStreamOperator<RowData, RowData> create(boolean
hasUpsertKey) {
+ return SinkUpsertMaterializerV2.create(
+ RowType.of(types),
+ equalizer(),
+ hasUpsertKey ? upsertEqualizer() : null,
+ hashFunction(),
+ hasUpsertKey ? upsertHashFunction() : null,
+ upsertKey(hasUpsertKey),
+ SequencedMultiSetStateConfig.defaults(
+ TimeDomain.PROCESSING_TIME, TTL_CONFIG));
Review Comment:
I've added them, configured as `StreamExecSink` does with the default
adaptive thresholds. ADAPTIVE stays on VALUE at 1 row and on MAP at 500. That
doubles its nightly time to about 35 minutes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
