You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Zakelly (via GitHub)" <gi...@apache.org> on 2023/12/16 11:27:31 UTC
[PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Zakelly opened a new pull request, #83:
URL: https://github.com/apache/flink-benchmarks/pull/83
As the title says, This PR introduce TTL state based benchmarks. The newly added tests are basically the same as the previous tests, but have been streamlined because the running time is too long with so many parameters.
I also suggest we disable it in the daily run, or only enable it weekly in jenkins since this is time-consuming.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #83:
URL: https://github.com/apache/flink-benchmarks/pull/83#issuecomment-1858829899
@masteryhx @Myasuka would you please take a look? Many thanks~
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Posted by "Myasuka (via GitHub)" <gi...@apache.org>.
Myasuka commented on code in PR #83:
URL: https://github.com/apache/flink-benchmarks/pull/83#discussion_r1429199281
##########
src/main/java/org/apache/flink/state/benchmark/ttl/TtlListStateBenchmark.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.benchmark.ttl;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
+import org.apache.flink.state.benchmark.StateBenchmarkBase;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.infra.Blackhole;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+import org.openjdk.jmh.runner.options.VerboseMode;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.flink.state.benchmark.StateBackendBenchmarkUtils.applyToAllKeys;
+import static org.apache.flink.state.benchmark.StateBackendBenchmarkUtils.compactState;
+import static org.apache.flink.state.benchmark.StateBackendBenchmarkUtils.getListState;
+import static org.apache.flink.state.benchmark.StateBenchmarkConstants.listValueCount;
+import static org.apache.flink.state.benchmark.StateBenchmarkConstants.setupKeyCount;
+
+/** Implementation for list state benchmark testing. */
+public class TtlListStateBenchmark extends TtlStateBenchmarkBase {
+ private final String STATE_NAME = "listState";
+ private final ListStateDescriptor<Long> STATE_DESC =
+ configTtl(new ListStateDescriptor<>(STATE_NAME, Long.class));
+ private ListState<Long> listState;
+ private List<Long> dummyLists;
+
+ public static void main(String[] args) throws RunnerException {
+ Options opt =
+ new OptionsBuilder()
+ .verbosity(VerboseMode.NORMAL)
+ .include(".*" + TtlListStateBenchmark.class.getCanonicalName() + ".*")
+ .build();
+
+ new Runner(opt).run();
+ }
+
+ @Setup
+ public void setUp() throws Exception {
+ keyedStateBackend = createKeyedStateBackend();
+ listState = getListState(keyedStateBackend, STATE_DESC);
+ dummyLists = new ArrayList<>(listValueCount);
+ for (int i = 0; i < listValueCount; ++i) {
+ dummyLists.add(random.nextLong());
+ }
+ keyIndex = new AtomicInteger();
+ }
+
+ @Setup(Level.Iteration)
+ public void setUpPerIteration() throws Exception {
+ for (int i = 0; i < setupKeyCount; ++i) {
+ keyedStateBackend.setCurrentKey((long) i);
+ listState.add(random.nextLong());
+ }
+ // make sure only one sst file left, so all get invocation will access this single file,
+ // to prevent the spike caused by different key distribution in multiple sst files,
+ // the more access to the older sst file, the lower throughput will be.
+ if (keyedStateBackend instanceof RocksDBKeyedStateBackend) {
Review Comment:
I noticed that there existed so much duplicated code here, can how reduce this code via Generics or more abstract class?
##########
src/main/java/org/apache/flink/state/benchmark/ttl/TtlStateBenchmarkBase.java:
##########
@@ -0,0 +1,48 @@
+package org.apache.flink.state.benchmark.ttl;
+
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.state.benchmark.StateBenchmarkBase;
+import org.openjdk.jmh.annotations.Param;
+
+import java.util.concurrent.TimeUnit;
+
+/** The base class for state tests with ttl. */
+public class TtlStateBenchmarkBase extends StateBenchmarkBase {
+
+ /** The expired time of ttl. */
+ public enum ExpiredTimeOptions {
+
+ /** 5 seconds. */
+ Seconds5(5000),
Review Comment:
What's the performance/result change if we tune this expired-seconds configuration?
Say if we increase it to `10` seconds, or decrease it to `3` seconds?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #83:
URL: https://github.com/apache/flink-benchmarks/pull/83#discussion_r1429208690
##########
src/main/java/org/apache/flink/state/benchmark/ttl/TtlListStateBenchmark.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.benchmark.ttl;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
+import org.apache.flink.state.benchmark.StateBenchmarkBase;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.infra.Blackhole;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+import org.openjdk.jmh.runner.options.VerboseMode;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.flink.state.benchmark.StateBackendBenchmarkUtils.applyToAllKeys;
+import static org.apache.flink.state.benchmark.StateBackendBenchmarkUtils.compactState;
+import static org.apache.flink.state.benchmark.StateBackendBenchmarkUtils.getListState;
+import static org.apache.flink.state.benchmark.StateBenchmarkConstants.listValueCount;
+import static org.apache.flink.state.benchmark.StateBenchmarkConstants.setupKeyCount;
+
+/** Implementation for list state benchmark testing. */
+public class TtlListStateBenchmark extends TtlStateBenchmarkBase {
+ private final String STATE_NAME = "listState";
+ private final ListStateDescriptor<Long> STATE_DESC =
+ configTtl(new ListStateDescriptor<>(STATE_NAME, Long.class));
+ private ListState<Long> listState;
+ private List<Long> dummyLists;
+
+ public static void main(String[] args) throws RunnerException {
+ Options opt =
+ new OptionsBuilder()
+ .verbosity(VerboseMode.NORMAL)
+ .include(".*" + TtlListStateBenchmark.class.getCanonicalName() + ".*")
+ .build();
+
+ new Runner(opt).run();
+ }
+
+ @Setup
+ public void setUp() throws Exception {
+ keyedStateBackend = createKeyedStateBackend();
+ listState = getListState(keyedStateBackend, STATE_DESC);
+ dummyLists = new ArrayList<>(listValueCount);
+ for (int i = 0; i < listValueCount; ++i) {
+ dummyLists.add(random.nextLong());
+ }
+ keyIndex = new AtomicInteger();
+ }
+
+ @Setup(Level.Iteration)
+ public void setUpPerIteration() throws Exception {
+ for (int i = 0; i < setupKeyCount; ++i) {
+ keyedStateBackend.setCurrentKey((long) i);
+ listState.add(random.nextLong());
+ }
+ // make sure only one sst file left, so all get invocation will access this single file,
+ // to prevent the spike caused by different key distribution in multiple sst files,
+ // the more access to the older sst file, the lower throughput will be.
+ if (keyedStateBackend instanceof RocksDBKeyedStateBackend) {
Review Comment:
Actually, in order to prevent small changes from affecting the existing test results, I did not use class abstraction purposely. I suggest we introduce a refactor to the existing tests along with the [FLINK-33864](https://issues.apache.org/jira/browse/FLINK-33864), since it can greatly affect test results. WDYT?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Posted by "Myasuka (via GitHub)" <gi...@apache.org>.
Myasuka commented on code in PR #83:
URL: https://github.com/apache/flink-benchmarks/pull/83#discussion_r1429398494
##########
src/main/java/org/apache/flink/state/benchmark/ttl/TtlListStateBenchmark.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.benchmark.ttl;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
+import org.apache.flink.state.benchmark.StateBenchmarkBase;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.infra.Blackhole;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+import org.openjdk.jmh.runner.options.VerboseMode;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.flink.state.benchmark.StateBackendBenchmarkUtils.applyToAllKeys;
+import static org.apache.flink.state.benchmark.StateBackendBenchmarkUtils.compactState;
+import static org.apache.flink.state.benchmark.StateBackendBenchmarkUtils.getListState;
+import static org.apache.flink.state.benchmark.StateBenchmarkConstants.listValueCount;
+import static org.apache.flink.state.benchmark.StateBenchmarkConstants.setupKeyCount;
+
+/** Implementation for list state benchmark testing. */
+public class TtlListStateBenchmark extends TtlStateBenchmarkBase {
+ private final String STATE_NAME = "listState";
+ private final ListStateDescriptor<Long> STATE_DESC =
+ configTtl(new ListStateDescriptor<>(STATE_NAME, Long.class));
+ private ListState<Long> listState;
+ private List<Long> dummyLists;
+
+ public static void main(String[] args) throws RunnerException {
+ Options opt =
+ new OptionsBuilder()
+ .verbosity(VerboseMode.NORMAL)
+ .include(".*" + TtlListStateBenchmark.class.getCanonicalName() + ".*")
+ .build();
+
+ new Runner(opt).run();
+ }
+
+ @Setup
+ public void setUp() throws Exception {
+ keyedStateBackend = createKeyedStateBackend();
+ listState = getListState(keyedStateBackend, STATE_DESC);
+ dummyLists = new ArrayList<>(listValueCount);
+ for (int i = 0; i < listValueCount; ++i) {
+ dummyLists.add(random.nextLong());
+ }
+ keyIndex = new AtomicInteger();
+ }
+
+ @Setup(Level.Iteration)
+ public void setUpPerIteration() throws Exception {
+ for (int i = 0; i < setupKeyCount; ++i) {
+ keyedStateBackend.setCurrentKey((long) i);
+ listState.add(random.nextLong());
+ }
+ // make sure only one sst file left, so all get invocation will access this single file,
+ // to prevent the spike caused by different key distribution in multiple sst files,
+ // the more access to the older sst file, the lower throughput will be.
+ if (keyedStateBackend instanceof RocksDBKeyedStateBackend) {
Review Comment:
I'm not sure whether such refactor could impact the performance results significantly, could you please provide some data if possible? If so, I think we can refactor the tests with FLINK-33864.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #83:
URL: https://github.com/apache/flink-benchmarks/pull/83#issuecomment-1874802825
@Myasuka Kindly remind~
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Posted by "Myasuka (via GitHub)" <gi...@apache.org>.
Myasuka commented on code in PR #83:
URL: https://github.com/apache/flink-benchmarks/pull/83#discussion_r1435532465
##########
src/main/java/org/apache/flink/state/benchmark/ttl/TtlStateBenchmarkBase.java:
##########
@@ -3,46 +3,102 @@
import org.apache.flink.api.common.state.StateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.state.benchmark.StateBenchmarkBase;
import org.openjdk.jmh.annotations.Param;
+import java.lang.reflect.Field;
import java.util.concurrent.TimeUnit;
/** The base class for state tests with ttl. */
public class TtlStateBenchmarkBase extends StateBenchmarkBase {
+ private static final long initialTime = 1000000;
+
/** The expired time of ttl. */
public enum ExpiredTimeOptions {
- /** 5 seconds. */
- Seconds5(5000),
+ /** Expire 3 percent of the initial keys per iteration. */
+ ExpirePercent3PerIteration(3),
Review Comment:
Should be `Expire3PercentPerIteration(3)`
##########
src/main/java/org/apache/flink/state/benchmark/ttl/TtlStateBenchmarkBase.java:
##########
@@ -3,46 +3,102 @@
import org.apache.flink.api.common.state.StateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.state.benchmark.StateBenchmarkBase;
import org.openjdk.jmh.annotations.Param;
+import java.lang.reflect.Field;
import java.util.concurrent.TimeUnit;
/** The base class for state tests with ttl. */
public class TtlStateBenchmarkBase extends StateBenchmarkBase {
+ private static final long initialTime = 1000000;
+
/** The expired time of ttl. */
public enum ExpiredTimeOptions {
- /** 5 seconds. */
- Seconds5(5000),
+ /** Expire 3 percent of the initial keys per iteration. */
+ ExpirePercent3PerIteration(3),
/** never expired but enable the ttl. */
- MaxTime(Long.MAX_VALUE);
+ NeverExpired(0);
- private Time time;
- ExpiredTimeOptions(long mills) {
- time = Time.of(mills, TimeUnit.MILLISECONDS);
+ public long advanceTimePerIteration;
+ ExpiredTimeOptions(int expirePercentPerIteration) {
+ this.advanceTimePerIteration = initialTime * expirePercentPerIteration / 100;
}
}
- @Param({"Seconds5", "MaxTime"})
- protected ExpiredTimeOptions expiredTime;
+ @Param({"ExpirePercent3PerIteration", "NeverExpired"})
+ protected ExpiredTimeOptions expiredOption;
@Param({"OnCreateAndWrite", "OnReadAndWrite"})
protected StateTtlConfig.UpdateType updateType;
- @Param({"ReturnExpiredIfNotCleanedUp", "NeverReturnExpired"})
+ @Param({"NeverReturnExpired"})
protected StateTtlConfig.StateVisibility stateVisibility;
+ protected ControllableTtlTimeProvider timeProvider;
+
/** Configure the state descriptor with ttl. */
protected <T extends StateDescriptor<?, ?>> T configTtl(T stateDescriptor) {
StateTtlConfig ttlConfig =
- new StateTtlConfig.Builder(expiredTime.time)
+ new StateTtlConfig.Builder(Time.of(initialTime, TimeUnit.MILLISECONDS))
.setUpdateType(updateType)
.setStateVisibility(stateVisibility)
.build();
stateDescriptor.enableTimeToLive(ttlConfig);
return stateDescriptor;
}
+
+ protected KeyedStateBackend<Long> createKeyedStateBackend() throws Exception {
+ KeyedStateBackend<Long> originalBackend = super.createKeyedStateBackend();
+ timeProvider = new ControllableTtlTimeProvider();
+ if (originalBackend instanceof AbstractKeyedStateBackend) {
+ // TODO: Config TtlTimeProvider in proper way.
Review Comment:
I think we can introduce a `TtlTimeProvider` in Flink's interface.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Posted by "Myasuka (via GitHub)" <gi...@apache.org>.
Myasuka commented on code in PR #83:
URL: https://github.com/apache/flink-benchmarks/pull/83#discussion_r1435528571
##########
src/main/java/org/apache/flink/state/benchmark/ttl/TtlStateBenchmarkBase.java:
##########
@@ -0,0 +1,48 @@
+package org.apache.flink.state.benchmark.ttl;
+
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.state.benchmark.StateBenchmarkBase;
+import org.openjdk.jmh.annotations.Param;
+
+import java.util.concurrent.TimeUnit;
+
+/** The base class for state tests with ttl. */
+public class TtlStateBenchmarkBase extends StateBenchmarkBase {
+
+ /** The expired time of ttl. */
+ public enum ExpiredTimeOptions {
+
+ /** 5 seconds. */
+ Seconds5(5000),
Review Comment:
I like the idea to use expired percent instead of the specific time-to-live. BTW, why choose 3% as the predefined value?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Posted by "Myasuka (via GitHub)" <gi...@apache.org>.
Myasuka merged PR #83:
URL: https://github.com/apache/flink-benchmarks/pull/83
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #83:
URL: https://github.com/apache/flink-benchmarks/pull/83#issuecomment-1869926898
@Myasuka I fixed a compile error and now it's OK. Maybe this repo needs a CI process validating the compilation.
Please let me know if you have any other concern on this PR. Thanks.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #83:
URL: https://github.com/apache/flink-benchmarks/pull/83#discussion_r1435779742
##########
src/main/java/org/apache/flink/state/benchmark/ttl/TtlStateBenchmarkBase.java:
##########
@@ -3,46 +3,102 @@
import org.apache.flink.api.common.state.StateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.ttl.TtlTimeProvider;
import org.apache.flink.state.benchmark.StateBenchmarkBase;
import org.openjdk.jmh.annotations.Param;
+import java.lang.reflect.Field;
import java.util.concurrent.TimeUnit;
/** The base class for state tests with ttl. */
public class TtlStateBenchmarkBase extends StateBenchmarkBase {
+ private static final long initialTime = 1000000;
+
/** The expired time of ttl. */
public enum ExpiredTimeOptions {
- /** 5 seconds. */
- Seconds5(5000),
+ /** Expire 3 percent of the initial keys per iteration. */
+ ExpirePercent3PerIteration(3),
/** never expired but enable the ttl. */
- MaxTime(Long.MAX_VALUE);
+ NeverExpired(0);
- private Time time;
- ExpiredTimeOptions(long mills) {
- time = Time.of(mills, TimeUnit.MILLISECONDS);
+ public long advanceTimePerIteration;
+ ExpiredTimeOptions(int expirePercentPerIteration) {
+ this.advanceTimePerIteration = initialTime * expirePercentPerIteration / 100;
}
}
- @Param({"Seconds5", "MaxTime"})
- protected ExpiredTimeOptions expiredTime;
+ @Param({"ExpirePercent3PerIteration", "NeverExpired"})
+ protected ExpiredTimeOptions expiredOption;
@Param({"OnCreateAndWrite", "OnReadAndWrite"})
protected StateTtlConfig.UpdateType updateType;
- @Param({"ReturnExpiredIfNotCleanedUp", "NeverReturnExpired"})
+ @Param({"NeverReturnExpired"})
protected StateTtlConfig.StateVisibility stateVisibility;
+ protected ControllableTtlTimeProvider timeProvider;
+
/** Configure the state descriptor with ttl. */
protected <T extends StateDescriptor<?, ?>> T configTtl(T stateDescriptor) {
StateTtlConfig ttlConfig =
- new StateTtlConfig.Builder(expiredTime.time)
+ new StateTtlConfig.Builder(Time.of(initialTime, TimeUnit.MILLISECONDS))
.setUpdateType(updateType)
.setStateVisibility(stateVisibility)
.build();
stateDescriptor.enableTimeToLive(ttlConfig);
return stateDescriptor;
}
+
+ protected KeyedStateBackend<Long> createKeyedStateBackend() throws Exception {
+ KeyedStateBackend<Long> originalBackend = super.createKeyedStateBackend();
+ timeProvider = new ControllableTtlTimeProvider();
+ if (originalBackend instanceof AbstractKeyedStateBackend) {
+ // TODO: Config TtlTimeProvider in proper way.
Review Comment:
Sure, I make a PR for this https://github.com/apache/flink/pull/23985
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #83:
URL: https://github.com/apache/flink-benchmarks/pull/83#discussion_r1435780122
##########
src/main/java/org/apache/flink/state/benchmark/ttl/TtlStateBenchmarkBase.java:
##########
@@ -0,0 +1,48 @@
+package org.apache.flink.state.benchmark.ttl;
+
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.state.benchmark.StateBenchmarkBase;
+import org.openjdk.jmh.annotations.Param;
+
+import java.util.concurrent.TimeUnit;
+
+/** The base class for state tests with ttl. */
+public class TtlStateBenchmarkBase extends StateBenchmarkBase {
+
+ /** The expired time of ttl. */
+ public enum ExpiredTimeOptions {
+
+ /** 5 seconds. */
+ Seconds5(5000),
Review Comment:
There are 20 iterations in each test, 10 rounds warm-up and 10 rounds testing. If 3% is chosen, the testing will cover 70% down to 40% state size, which is a relatively reasonable value (not so big or small).
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #83:
URL: https://github.com/apache/flink-benchmarks/pull/83#discussion_r1430371694
##########
src/main/java/org/apache/flink/state/benchmark/ttl/TtlListStateBenchmark.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.benchmark.ttl;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
+import org.apache.flink.state.benchmark.StateBenchmarkBase;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.infra.Blackhole;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+import org.openjdk.jmh.runner.options.VerboseMode;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.flink.state.benchmark.StateBackendBenchmarkUtils.applyToAllKeys;
+import static org.apache.flink.state.benchmark.StateBackendBenchmarkUtils.compactState;
+import static org.apache.flink.state.benchmark.StateBackendBenchmarkUtils.getListState;
+import static org.apache.flink.state.benchmark.StateBenchmarkConstants.listValueCount;
+import static org.apache.flink.state.benchmark.StateBenchmarkConstants.setupKeyCount;
+
+/** Implementation for list state benchmark testing. */
+public class TtlListStateBenchmark extends TtlStateBenchmarkBase {
+ private final String STATE_NAME = "listState";
+ private final ListStateDescriptor<Long> STATE_DESC =
+ configTtl(new ListStateDescriptor<>(STATE_NAME, Long.class));
+ private ListState<Long> listState;
+ private List<Long> dummyLists;
+
+ public static void main(String[] args) throws RunnerException {
+ Options opt =
+ new OptionsBuilder()
+ .verbosity(VerboseMode.NORMAL)
+ .include(".*" + TtlListStateBenchmark.class.getCanonicalName() + ".*")
+ .build();
+
+ new Runner(opt).run();
+ }
+
+ @Setup
+ public void setUp() throws Exception {
+ keyedStateBackend = createKeyedStateBackend();
+ listState = getListState(keyedStateBackend, STATE_DESC);
+ dummyLists = new ArrayList<>(listValueCount);
+ for (int i = 0; i < listValueCount; ++i) {
+ dummyLists.add(random.nextLong());
+ }
+ keyIndex = new AtomicInteger();
+ }
+
+ @Setup(Level.Iteration)
+ public void setUpPerIteration() throws Exception {
+ for (int i = 0; i < setupKeyCount; ++i) {
+ keyedStateBackend.setCurrentKey((long) i);
+ listState.add(random.nextLong());
+ }
+ // make sure only one sst file left, so all get invocation will access this single file,
+ // to prevent the spike caused by different key distribution in multiple sst files,
+ // the more access to the older sst file, the lower throughput will be.
+ if (keyedStateBackend instanceof RocksDBKeyedStateBackend) {
Review Comment:
After some try, I found there is no easy way to do so, since the `Ttl*StateBenchmark` extends the `TtlStateBenchmarkBase`, while they cannot extends `*StateBenchmark` in the meantime. So a possible way is to extract an `TestHarness` or `TestContext` to hold the variables. I only do refactor in `valueAdd`, and test it on my local machine. Before:
```
Benchmark (backendType) Mode Cnt Score Error Units
ValueStateBenchmark.valueAdd HEAP thrpt 30 5815.682 ± 122.008 ops/ms
Benchmark (backendType) Mode Cnt Score Error Units
ValueStateBenchmark.valueAdd HEAP thrpt 30 5817.108 ± 162.588 ops/ms
Benchmark (backendType) Mode Cnt Score Error Units
ValueStateBenchmark.valueAdd HEAP thrpt 30 5792.671 ± 132.356 ops/ms
```
And after the refactor:
```
Benchmark (backendType) Mode Cnt Score Error Units
ValueStateBenchmark.valueAdd HEAP thrpt 30 5712.225 ± 133.996 ops/ms
Benchmark (backendType) Mode Cnt Score Error Units
ValueStateBenchmark.valueAdd HEAP thrpt 30 5733.398 ± 77.725 ops/ms
Benchmark (backendType) Mode Cnt Score Error Units
ValueStateBenchmark.valueAdd HEAP thrpt 30 5771.855 ± 80.658 ops/ms
```
As we can see, there is about 1% impact on the performance results.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #83:
URL: https://github.com/apache/flink-benchmarks/pull/83#discussion_r1429207983
##########
src/main/java/org/apache/flink/state/benchmark/ttl/TtlStateBenchmarkBase.java:
##########
@@ -0,0 +1,48 @@
+package org.apache.flink.state.benchmark.ttl;
+
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.state.benchmark.StateBenchmarkBase;
+import org.openjdk.jmh.annotations.Param;
+
+import java.util.concurrent.TimeUnit;
+
+/** The base class for state tests with ttl. */
+public class TtlStateBenchmarkBase extends StateBenchmarkBase {
+
+ /** The expired time of ttl. */
+ public enum ExpiredTimeOptions {
+
+ /** 5 seconds. */
+ Seconds5(5000),
Review Comment:
Considering that the warm-up will take `10` seconds, `5` seconds is enough for the test to enter the key expiration phase, and it is not too short. I will give a comparison of the results under different configurations.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #83:
URL: https://github.com/apache/flink-benchmarks/pull/83#discussion_r1430468167
##########
src/main/java/org/apache/flink/state/benchmark/ttl/TtlStateBenchmarkBase.java:
##########
@@ -0,0 +1,48 @@
+package org.apache.flink.state.benchmark.ttl;
+
+import org.apache.flink.api.common.state.StateDescriptor;
+import org.apache.flink.api.common.state.StateTtlConfig;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.state.benchmark.StateBenchmarkBase;
+import org.openjdk.jmh.annotations.Param;
+
+import java.util.concurrent.TimeUnit;
+
+/** The base class for state tests with ttl. */
+public class TtlStateBenchmarkBase extends StateBenchmarkBase {
+
+ /** The expired time of ttl. */
+ public enum ExpiredTimeOptions {
+
+ /** 5 seconds. */
+ Seconds5(5000),
Review Comment:
I think it is better to manipulate the `TtlTimeProvider` to finely control the number of keys eliminated in each iteration. So I customize the `TtlTimeProvider` and provide an option for the percentage of keys that expired per iteration. This option will affect the result of test especially in `valueGet`. Here's a running test:
```
# Benchmark: org.apache.flink.state.benchmark.ttl.TtlValueStateBenchmark.valueGet
# Parameters: (backendType = HEAP, expiredOption = ExpirePercent3PerIteration, stateVisibility = NeverReturnExpired, updateType = OnCreateAndWrite)
# Run progress: 0.00% complete, ETA 00:12:00
# Fork: 1 of 3
# Warmup Iteration 1: 2993.246 ops/ms
# Warmup Iteration 2: 3520.954 ops/ms
# Warmup Iteration 3: 3640.884 ops/ms
# Warmup Iteration 4: 3643.386 ops/ms
# Warmup Iteration 5: 3776.677 ops/ms
# Warmup Iteration 6: 3739.375 ops/ms
# Warmup Iteration 7: 3903.434 ops/ms
# Warmup Iteration 8: 3877.286 ops/ms
# Warmup Iteration 9: 3913.663 ops/ms
# Warmup Iteration 10: 4093.471 ops/ms
Iteration 1: 4117.018 ops/ms
Iteration 2: 4255.166 ops/ms
Iteration 3: 4338.564 ops/ms
Iteration 4: 4439.571 ops/ms
Iteration 5: 4603.235 ops/ms
Iteration 6: 4735.039 ops/ms
Iteration 7: 4896.946 ops/ms
Iteration 8: 5153.644 ops/ms
Iteration 9: 5429.196 ops/ms
Iteration 10: 5772.806 ops/ms
```
As we can see, as the key expires, the test results increase iteration by iteration. I pick 3% for a predefined value, since there are 20 iterations in each test and we cannot let all the keys expire. WDYT? @Myasuka
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-30535] Introduce TTL state based benchmarks [flink-benchmarks]
Posted by "Myasuka (via GitHub)" <gi...@apache.org>.
Myasuka commented on code in PR #83:
URL: https://github.com/apache/flink-benchmarks/pull/83#discussion_r1435531688
##########
src/main/java/org/apache/flink/state/benchmark/ttl/TtlListStateBenchmark.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.benchmark.ttl;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend;
+import org.apache.flink.state.benchmark.StateBenchmarkBase;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.infra.Blackhole;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+import org.openjdk.jmh.runner.options.VerboseMode;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.flink.state.benchmark.StateBackendBenchmarkUtils.applyToAllKeys;
+import static org.apache.flink.state.benchmark.StateBackendBenchmarkUtils.compactState;
+import static org.apache.flink.state.benchmark.StateBackendBenchmarkUtils.getListState;
+import static org.apache.flink.state.benchmark.StateBenchmarkConstants.listValueCount;
+import static org.apache.flink.state.benchmark.StateBenchmarkConstants.setupKeyCount;
+
+/** Implementation for list state benchmark testing. */
+public class TtlListStateBenchmark extends TtlStateBenchmarkBase {
+ private final String STATE_NAME = "listState";
+ private final ListStateDescriptor<Long> STATE_DESC =
+ configTtl(new ListStateDescriptor<>(STATE_NAME, Long.class));
+ private ListState<Long> listState;
+ private List<Long> dummyLists;
+
+ public static void main(String[] args) throws RunnerException {
+ Options opt =
+ new OptionsBuilder()
+ .verbosity(VerboseMode.NORMAL)
+ .include(".*" + TtlListStateBenchmark.class.getCanonicalName() + ".*")
+ .build();
+
+ new Runner(opt).run();
+ }
+
+ @Setup
+ public void setUp() throws Exception {
+ keyedStateBackend = createKeyedStateBackend();
+ listState = getListState(keyedStateBackend, STATE_DESC);
+ dummyLists = new ArrayList<>(listValueCount);
+ for (int i = 0; i < listValueCount; ++i) {
+ dummyLists.add(random.nextLong());
+ }
+ keyIndex = new AtomicInteger();
+ }
+
+ @Setup(Level.Iteration)
+ public void setUpPerIteration() throws Exception {
+ for (int i = 0; i < setupKeyCount; ++i) {
+ keyedStateBackend.setCurrentKey((long) i);
+ listState.add(random.nextLong());
+ }
+ // make sure only one sst file left, so all get invocation will access this single file,
+ // to prevent the spike caused by different key distribution in multiple sst files,
+ // the more access to the older sst file, the lower throughput will be.
+ if (keyedStateBackend instanceof RocksDBKeyedStateBackend) {
Review Comment:
I think the 1% impact is really small compared to the project maintaining. And we can postpone this later in [FLINK-33864](https://issues.apache.org/jira/browse/FLINK-33864).
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org