You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/07/15 09:59:20 UTC
[GitHub] [flink-benchmarks] fredia opened a new pull request #23: [FLINK-23399][runtime/state]Add a benchmark for rescaling
fredia opened a new pull request #23:
URL: https://github.com/apache/flink-benchmarks/pull/23
This pull request adds a benchmark for rescaling.
The benchmark process is:
(1) generate some states,
(2) change the parallelism of the operator, and restore from these states generate before.
The implementation of this benchmark is based on `RocksIncrementalCheckpointRescalingTest`, and **Average Time** is used to measure the rescaling performance on each subtask.
And the results of this benchmark are:
```text
Benchmark (backendType) (parallelism1) (parallelism2) (subtaskIndex) Mode Cnt Score Error Units
RocksIncrementalCheckpointReScaleBenchmark.rescale heap 3 4 0 avgt 3 17.326 ± 3.303 ms/op
RocksIncrementalCheckpointReScaleBenchmark.rescale heap 3 4 1 avgt 3 14.874 ± 17.775 ms/op
RocksIncrementalCheckpointReScaleBenchmark.rescale heap 3 4 2 avgt 3 18.429 ± 13.515 ms/op
RocksIncrementalCheckpointReScaleBenchmark.rescale heap 3 4 3 avgt 3 13.652 ± 8.159 ms/op
RocksIncrementalCheckpointReScaleBenchmark.rescale fileSystem 3 4 0 avgt 3 17.478 ± 3.431 ms/op
RocksIncrementalCheckpointReScaleBenchmark.rescale fileSystem 3 4 1 avgt 3 13.762 ± 8.742 ms/op
RocksIncrementalCheckpointReScaleBenchmark.rescale fileSystem 3 4 2 avgt 3 17.908 ± 10.943 ms/op
RocksIncrementalCheckpointReScaleBenchmark.rescale fileSystem 3 4 3 avgt 3 14.823 ± 13.521 ms/op
RocksIncrementalCheckpointReScaleBenchmark.rescale rocksdb 3 4 0 avgt 3 108.648 ± 24.771 ms/op
RocksIncrementalCheckpointReScaleBenchmark.rescale rocksdb 3 4 1 avgt 3 83.095 ± 60.409 ms/op
RocksIncrementalCheckpointReScaleBenchmark.rescale rocksdb 3 4 2 avgt 3 108.561 ± 36.595 ms/op
RocksIncrementalCheckpointReScaleBenchmark.rescale rocksdb 3 4 3 avgt 3 83.962 ± 20.501 ms/op
RocksIncrementalCheckpointReScaleBenchmark.rescale rocksdbIncr 3 4 0 avgt 3 188.484 ± 16.489 ms/op
RocksIncrementalCheckpointReScaleBenchmark.rescale rocksdbIncr 3 4 1 avgt 3 276.437 ± 20.479 ms/op
RocksIncrementalCheckpointReScaleBenchmark.rescale rocksdbIncr 3 4 2 avgt 3 293.488 ± 156.933 ms/op
RocksIncrementalCheckpointReScaleBenchmark.rescale rocksdbIncr 3 4 3 avgt 3 182.925 ± 14.381 ms/op
```
--
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
[GitHub] [flink-benchmarks] pnowojski commented on pull request #23: [FLINK-23399][runtime/state]Add a benchmark for rescaling
Posted by GitBox <gi...@apache.org>.
pnowojski commented on pull request #23:
URL: https://github.com/apache/flink-benchmarks/pull/23#issuecomment-880602174
Thanks for contributing those. Couple of remarks:
1. This code is using Flink non public API, so please take a look at the [README](https://github.com/apache/flink-benchmarks#code-structure) (I've updated it a second ago to give more details/pointers)
2. The benchmarks seems quite unstable: `18.429 ± 13.515 ms/op`. That probably makes them not very useful :( Could you try to rewrite it somehow? Maybe include more rescaling operations in one invocation?
3. What's the total running time of the newly added benchmarks? Maybe we could avoid testing `heap` parameter and just focus on single `subtaskIndex` (the middle one?)?
--
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
[GitHub] [flink-benchmarks] pnowojski commented on pull request #23: [FLINK-23399][runtime/state]Add a benchmark for rescaling
Posted by GitBox <gi...@apache.org>.
pnowojski commented on pull request #23:
URL: https://github.com/apache/flink-benchmarks/pull/23#issuecomment-884219160
1. It looks like [the place you mentioned](https://github.com/apache/flink/tree/master/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark) is a good one to start
About the 30 minutes run. Yes, dropping `heap` and picking a single value for `subTaskIndex` would help a lot to bring this time down. Also keep in mind the recommendation from the readme:
> Rule of thumb is that you should increase the number of processed records up to a point where results stop improving visibly, while trying to keep the single benchmark invocation around/under 1 second.
With single iteration 1 second, the total time should get down to ~4minutes. But there can be a trade off with too short single invocation, the benchmarks can be unstable.
--
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
[GitHub] [flink-benchmarks] fredia commented on pull request #23: [FLINK-23399][state] Add a benchmark for rescaling
Posted by GitBox <gi...@apache.org>.
fredia commented on pull request #23:
URL: https://github.com/apache/flink-benchmarks/pull/23#issuecomment-1086559519
hi @pnowojski, I updated this PR according to the suggestion above, would you please take a review again?
I move some code to [flink-state-backends](https://github.com/apache/flink/tree/master/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark) in https://github.com/apache/flink/pull/19177, and split the previous RescalingBenchmark into `RocksdbStateBackendRescalingBenchmark` and `FsStateBackendRescalingBenchmark`.
The results of those benchmarks are:
```
# Run complete. Total time: 00:18:53
Benchmark (rescaleType) Mode Cnt Score Error Units
RocksdbStateBackendRescalingBenchmarkExecutor.rescale RESCALE_IN avgt 30 19825.918 ± 1261.372 ms/op
RocksdbStateBackendRescalingBenchmarkExecutor.rescale RESCALE_OUT avgt 30 1811.478 ± 82.262 ms/op
```
```
# Run complete. Total time: 00:17:11
Benchmark (rescaleType) Mode Cnt Score Error Units
FsStateBackendRescalingBenchmarkExecutor.rescale RESCALE_IN avgt 30 15890.784 ± 461.543 ms/op
FsStateBackendBenchmarkExecutor.rescale RESCALE_OUT avgt 30 7858.031 ± 276.433 ms/op
```
In practice, most rescaling scenarios are recovering from a large state(>=1GB). So I set the state to be larger, the rule of “keep the single benchmark invocation around/under 1 second” is not satisfied, but the total run time of each benchmark is under 30 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-benchmarks] fredia edited a comment on pull request #23: [FLINK-23399][state] Add a benchmark for rescaling
Posted by GitBox <gi...@apache.org>.
fredia edited a comment on pull request #23:
URL: https://github.com/apache/flink-benchmarks/pull/23#issuecomment-1086559519
hi @pnowojski, I updated this PR according to the suggestion above, would you please take a review again?
I move some code to [flink-state-backends](https://github.com/apache/flink/tree/master/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark) in https://github.com/apache/flink/pull/19177, and split the previous RescalingBenchmark into `RocksdbStateBackendRescalingBenchmark` and `FsStateBackendRescalingBenchmark`.
The results of those benchmarks are:
```
# Run complete. Total time: 00:18:53
Benchmark (rescaleType) Mode Cnt Score Error Units
RocksdbStateBackendRescalingBenchmarkExecutor.rescale RESCALE_IN avgt 30 19825.918 ± 1261.372 ms/op
RocksdbStateBackendRescalingBenchmarkExecutor.rescale RESCALE_OUT avgt 30 1811.478 ± 82.262 ms/op
```
```
# Run complete. Total time: 00:17:11
Benchmark (rescaleType) Mode Cnt Score Error Units
FsStateBackendRescalingBenchmarkExecutor.rescale RESCALE_IN avgt 30 15890.784 ± 461.543 ms/op
FsStateBackendRescalingBenchmarkExecutor.rescale RESCALE_OUT avgt 30 7858.031 ± 276.433 ms/op
```
In practice, most rescaling scenarios are recovering from a large state(>=1GB). So I set the state to be larger, the rule of “keep the single benchmark invocation around/under 1 second” is not satisfied, but the total run time of each benchmark is under 30 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-benchmarks] fredia edited a comment on pull request #23: [FLINK-23399][runtime/state]Add a benchmark for rescaling
Posted by GitBox <gi...@apache.org>.
fredia edited a comment on pull request #23:
URL: https://github.com/apache/flink-benchmarks/pull/23#issuecomment-882302478
Thanks for your reply and nice suggestions, @pnowojski.
1. Yes, this code uses some `StreamOperatorTestHarness` which are not public. According to the code structure suggestions, we would move code to [flink-state-backends](https://github.com/apache/flink/tree/master/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark), but not sure if it is appropriate, could you give some suggestions on where to move?
2. Sorry for the careless results, I only set `iterations=1` and `fork value=3` before. (。﹏。*)
I increase iterations to 10 and re-run the test, the new results are shown below.
And the state size can also affect the results, the larger the state size, the more stable the results. But the maximum state size is limit by memory-backed, we would remove the `heap` parameter afterward.
```text
Benchmark (backendType) (parallelism1) (parallelism2) (subtaskIndex) Mode Cnt Score Error Units
RescalingBenchmark.rescale heap 3 4 0 avgt 30 8.557 ± 0.501 ms/op
RescalingBenchmark.rescale heap 3 4 1 avgt 30 6.405 ± 0.217 ms/op
RescalingBenchmark.rescale heap 3 4 2 avgt 30 8.446 ± 0.303 ms/op
RescalingBenchmark.rescale heap 3 4 3 avgt 30 6.951 ± 0.634 ms/op
RescalingBenchmark.rescale fileSystem 3 4 0 avgt 30 8.967 ± 0.508 ms/op
RescalingBenchmark.rescale fileSystem 3 4 1 avgt 30 6.615 ± 0.273 ms/op
RescalingBenchmark.rescale fileSystem 3 4 2 avgt 30 8.833 ± 0.472 ms/op
RescalingBenchmark.rescale fileSystem 3 4 3 avgt 30 6.737 ± 0.302 ms/op
RescalingBenchmark.rescale rocksdb 3 4 0 avgt 30 100.949 ± 1.829 ms/op
RescalingBenchmark.rescale rocksdb 3 4 1 avgt 30 82.988 ± 6.050 ms/op
RescalingBenchmark.rescale rocksdb 3 4 2 avgt 30 104.332 ± 2.500 ms/op
RescalingBenchmark.rescale rocksdb 3 4 3 avgt 30 78.468 ± 1.353 ms/op
RescalingBenchmark.rescale rocksdbIncr 3 4 0 avgt 30 184.872 ± 4.912 ms/op
RescalingBenchmark.rescale rocksdbIncr 3 4 1 avgt 30 240.067 ± 7.431 ms/op
RescalingBenchmark.rescale rocksdbIncr 3 4 2 avgt 30 242.117 ± 3.595 ms/op
RescalingBenchmark.rescale rocksdbIncr 3 4 3 avgt 30 150.236 ± 2.395 ms/op
```
3. The total running time is about 30 min when `warmup iterations = 10`, `measurement iterations = 10`, and `fork value=3`. Agree to avoid testing heap parameter, we would remove the heap parameter later.
And we plan to pick out some typical cases(such as restore from partial state handle, restore from multiple state handles etc.) instead of testing all `subtaskIndex`.
Thank you for the great suggestions. I'll improve my codes according to these.
--
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
[GitHub] [flink-benchmarks] fredia commented on pull request #23: [FLINK-23399][runtime/state]Add a benchmark for rescaling
Posted by GitBox <gi...@apache.org>.
fredia commented on pull request #23:
URL: https://github.com/apache/flink-benchmarks/pull/23#issuecomment-882302478
Thanks for your reply and nice suggestions，
1. Yes, this code use some `StreamOperatorTestHarness` which are not public. According to the code structure suggestions, we would move code to [flink-state-backends](https://github.com/apache/flink/tree/master/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark), but not sure if it is appropriate, could you give some suggestions on where to move?
2. Sorry for the careless results, I only set `iterations=1` and `fork value=3` before. (。﹏。*)
I increase iterations to 10 and re-run the test, the new results are shown below.
And the state size can also affect the results, the larger the state size, the more stable the results. But the maximum state size is limit by memory-backed, we would remove the `heap` parameter afterward.
```text
Benchmark (backendType) (parallelism1) (parallelism2) (subtaskIndex) Mode Cnt Score Error Units
RescalingBenchmark.rescale heap 3 4 0 avgt 30 8.557 ± 0.501 ms/op
RescalingBenchmark.rescale heap 3 4 1 avgt 30 6.405 ± 0.217 ms/op
RescalingBenchmark.rescale heap 3 4 2 avgt 30 8.446 ± 0.303 ms/op
RescalingBenchmark.rescale heap 3 4 3 avgt 30 6.951 ± 0.634 ms/op
RescalingBenchmark.rescale fileSystem 3 4 0 avgt 30 8.967 ± 0.508 ms/op
RescalingBenchmark.rescale fileSystem 3 4 1 avgt 30 6.615 ± 0.273 ms/op
RescalingBenchmark.rescale fileSystem 3 4 2 avgt 30 8.833 ± 0.472 ms/op
RescalingBenchmark.rescale fileSystem 3 4 3 avgt 30 6.737 ± 0.302 ms/op
RescalingBenchmark.rescale rocksdb 3 4 0 avgt 30 100.949 ± 1.829 ms/op
RescalingBenchmark.rescale rocksdb 3 4 1 avgt 30 82.988 ± 6.050 ms/op
RescalingBenchmark.rescale rocksdb 3 4 2 avgt 30 104.332 ± 2.500 ms/op
RescalingBenchmark.rescale rocksdb 3 4 3 avgt 30 78.468 ± 1.353 ms/op
RescalingBenchmark.rescale rocksdbIncr 3 4 0 avgt 30 184.872 ± 4.912 ms/op
RescalingBenchmark.rescale rocksdbIncr 3 4 1 avgt 30 240.067 ± 7.431 ms/op
RescalingBenchmark.rescale rocksdbIncr 3 4 2 avgt 30 242.117 ± 3.595 ms/op
RescalingBenchmark.rescale rocksdbIncr 3 4 3 avgt 30 150.236 ± 2.395 ms/op
```
3. The total running time is about 30 min when `warmup iterations = 10`, `measurement iterations = 10`, and `fork value=3`. Agree to avoid testing heap parameter, we would remove the heap parameter later.
And we plan to pick out some typical cases(such as restore from partial state handle, restore from multiple state handles etc.) instead of testing all `subtaskIndex`.
Thank you for the great suggestions. I'll improve my codes according to these.
--
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
[GitHub] [flink-benchmarks] fredia commented on pull request #23: [FLINK-23399][state] Add a benchmark for rescaling
Posted by GitBox <gi...@apache.org>.
fredia commented on pull request #23:
URL: https://github.com/apache/flink-benchmarks/pull/23#issuecomment-1086609117
Thanks for the feedback! @Myasuka
The configurations and results for rocksdb state backend are:
<div class="lake-content" typography="classic">
| rescale type | delete type | overlap threshold | time
-- | -- | -- | -- | --
1 | 1->2 | native deleteRange | 0 | 1811.478 ± 82.262 ms/op
2 | 1->2 | native deleteRange | 0.75 | 17640.542 ± 1784.368 ms/op
3 | 1->2 | scan and delete | 0 | 21790.634 ± 216.322 ms/op
4 | 1->2 | scan and delete | 0.75 | 19595.801 ± 523.404 ms/op
5 | 2->1 | native deleteRange | 0 | 19825.918 ± 1261.372 ms/op
6 | 2->1 | native deleteRange | 0.75 | 20121.673 ± 1298.390 ms/op
7 | 2->1 | scan and delete | 0 | 18628.801 ± 289.178 ms/op
8 | 2->1 | scan and delete | 0.75 | 18520.173 ± 616.218 ms/op
</div>
For RocksDB state backend:
For rescale-in cases(line5~line8), the actual overlap fraction is 1.0, regardless of whether the threshold is 0 or 0.75, one handle is selected as the init handle to initialize the base DB, and then another handle is put into the base DB one by one. There is nothing to delete, using `scan and delete` or `native deleteRange` has no effects. So, we can see the results in line5 to line8 are about the same.
For rescale-out cases(line1~line4), the actual overlap fraction is 0.5,
- if threshold=0, the handle(only one) is selected as the init handle to initialize the base DB, and then half of the handle would be deleted. In this situation, `native deleteRange` is better than `scan and delete`(line1 and line3), this also proves the performance improvement of [FLINK-21321](https://issues.apache.org/jira/browse/FLINK-21321).
- if threshold=0.75, it would build an empty DB as base DB, and put half of the handle into base DB, thus using `scan and delete` or `native deleteRange` has no effects(line2 and line4).
For file system state backend, I set fileStateSizeThreshold=0 and got the results:
```
# Run complete. Total time: 00:17:11
Benchmark (rescaleType) Mode Cnt Score Error Units
FsStateBackendRescalingBenchmarkExecutor.rescale RESCALE_IN avgt 30 15890.784 ± 461.543 ms/op
FsStateBackendRescalingBenchmarkExecutor.rescale RESCALE_OUT avgt 30 7858.031 ± 276.433 ms/op
```
--
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
[GitHub] [flink-benchmarks] Myasuka commented on pull request #23: [FLINK-23399][state] Add a benchmark for rescaling
Posted by GitBox <gi...@apache.org>.
Myasuka commented on pull request #23:
URL: https://github.com/apache/flink-benchmarks/pull/23#issuecomment-1086577485
@fredia Would you please share the configuration for each state backend test?
And we can give some analysis on the performance behavior of different state-backends.
For RocksDB state backend:
the rescale-in case need to scan one DB and insert to the init DB.
the rescale-out case need to call delete range on the init DB (after deleteRange feature merged in FLINK-21321).
The process of scan and insert is certainly much longer than writing delete range.
For file system state backend (we make the original data size as the same):
the rescale-in case need to fill up 100% data into one subtask.
the rescale-out case need to fill up 50% data into one subtask. That's why we see the result number of rescale-in is twice as rescale-out.
If could separate the benchmark on checkpoint rescaling from other benchmarks, the longger operation duration would not impact much.
BTW, could you also share the result to prove the performance improvement of FLINK-21321?
--
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
[GitHub] [flink-benchmarks] Myasuka commented on a change in pull request #23: [FLINK-23399][state] Add a benchmark for rescaling
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #23:
URL: https://github.com/apache/flink-benchmarks/pull/23#discussion_r841035064
##########
File path: src/main/java/org/apache/flink/checkpoint/benchmark/FsStateBackendRescalingBenchmarkExecutor.java
##########
@@ -0,0 +1,88 @@
+/*
+ * 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.checkpoint.benchmark;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.config.ConfigUtil;
+import org.apache.flink.config.StateBenchmarkOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.contrib.streaming.state.benchmark.RescalingBenchmarkBuilder;
+import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
+import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
+import org.openjdk.jmh.annotations.*;
+import org.openjdk.jmh.runner.RunnerException;
+
+import java.io.IOException;
+import java.net.URI;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.openjdk.jmh.annotations.Mode.AverageTime;
+
+@OutputTimeUnit(MILLISECONDS)
+@BenchmarkMode(AverageTime)
+@Warmup(iterations = 3)
+public class FsStateBackendRescalingBenchmarkExecutor extends RescalingBenchmarkBase {
+ private final int numberOfKeys = 2500000;
Review comment:
Could you share the state size with your current configuration, and add it as the comments.
This is also true for `RocksdbStateBackendRescalingBenchmarkExecutor`
##########
File path: src/main/java/org/apache/flink/checkpoint/benchmark/RescalingBenchmarkBase.java
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.checkpoint.benchmark;
+
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.benchmark.BenchmarkBase;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.contrib.streaming.state.benchmark.RescalingBenchmark;
+import org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadLocalRandom;
+import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.Collector;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+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.Arrays;
+import java.util.Iterator;
+
+public class RescalingBenchmarkBase extends BenchmarkBase {
+
+ @Param({"RESCALE_IN", "RESCALE_OUT"})
+ protected RescaleType rescaleType;
+
+ protected RescalingBenchmark<byte[]> benchmark;
+
+ public static void runBenchmark(Class<?> clazz) throws RunnerException {
+ Options options =
+ new OptionsBuilder()
+ .verbosity(VerboseMode.NORMAL)
+ .include(".*" + clazz.getCanonicalName() + ".*")
+ .build();
+
+ new Runner(options).run();
+ }
+
+ @State(Scope.Thread)
+ public enum RescaleType {
+ RESCALE_OUT(1, 2, 0),
+ RESCALE_IN(2, 1, 0);
+
+ private final int parallelismBefore;
+ private final int parallelismAfter;
+ private final int subtaskIndex;
+
+ RescaleType(int parallelismBefore, int parallelismAfter, int subtaskIdx) {
+ this.parallelismBefore = parallelismBefore;
+ this.parallelismAfter = parallelismAfter;
+ this.subtaskIndex = subtaskIdx;
+ }
+
+ public int getParallelismBefore() {
+ return parallelismBefore;
+ }
+
+ public int getParallelismAfter() {
+ return parallelismAfter;
+ }
+
+ public int getSubtaskIndex() {
+ return subtaskIndex;
+ }
+ }
+
+ protected static class ByteArrayRecordGenerator
+ implements RescalingBenchmark.StreamRecordGenerator<byte[]> {
+ private final int numberOfKeys;
+ private final byte[] fatArray;
+ private int count = 0;
+
+ protected ByteArrayRecordGenerator(final int numberOfKeys,
+ final int keyLen) {
+ this.numberOfKeys = numberOfKeys;
+ fatArray = new byte[keyLen];
+ }
+
+ // generate deterministic elements for source
+ @Override
+ public Iterator<StreamRecord<byte[]>> generate() {
+ return new Iterator<StreamRecord<byte[]>>() {
+ @Override
+ public boolean hasNext() {
+ return count < numberOfKeys;
+ }
+
+ @Override
+ public StreamRecord<byte[]> next() {
+ ThreadLocalRandom.current().nextBytes(fatArray);
Review comment:
This random cannot guarantee deterministic source.
##########
File path: src/main/java/org/apache/flink/checkpoint/benchmark/RescalingBenchmarkBase.java
##########
@@ -0,0 +1,161 @@
+/*
+ * 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.checkpoint.benchmark;
+
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.benchmark.BenchmarkBase;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.contrib.streaming.state.benchmark.RescalingBenchmark;
+import org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadLocalRandom;
+import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.Collector;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+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.Arrays;
+import java.util.Iterator;
+
+public class RescalingBenchmarkBase extends BenchmarkBase {
+
+ @Param({"RESCALE_IN", "RESCALE_OUT"})
+ protected RescaleType rescaleType;
+
+ protected RescalingBenchmark<byte[]> benchmark;
+
+ public static void runBenchmark(Class<?> clazz) throws RunnerException {
+ Options options =
+ new OptionsBuilder()
+ .verbosity(VerboseMode.NORMAL)
+ .include(".*" + clazz.getCanonicalName() + ".*")
+ .build();
+
+ new Runner(options).run();
+ }
+
+ @State(Scope.Thread)
+ public enum RescaleType {
+ RESCALE_OUT(1, 2, 0),
+ RESCALE_IN(2, 1, 0);
+
+ private final int parallelismBefore;
+ private final int parallelismAfter;
+ private final int subtaskIndex;
+
+ RescaleType(int parallelismBefore, int parallelismAfter, int subtaskIdx) {
+ this.parallelismBefore = parallelismBefore;
+ this.parallelismAfter = parallelismAfter;
+ this.subtaskIndex = subtaskIdx;
+ }
+
+ public int getParallelismBefore() {
+ return parallelismBefore;
+ }
+
+ public int getParallelismAfter() {
+ return parallelismAfter;
+ }
+
+ public int getSubtaskIndex() {
+ return subtaskIndex;
+ }
+ }
+
+ protected static class ByteArrayRecordGenerator
+ implements RescalingBenchmark.StreamRecordGenerator<byte[]> {
+ private final int numberOfKeys;
+ private final byte[] fatArray;
+ private int count = 0;
+
+ protected ByteArrayRecordGenerator(final int numberOfKeys,
+ final int keyLen) {
+ this.numberOfKeys = numberOfKeys;
+ fatArray = new byte[keyLen];
+ }
+
+ // generate deterministic elements for source
+ @Override
+ public Iterator<StreamRecord<byte[]>> generate() {
+ return new Iterator<StreamRecord<byte[]>>() {
+ @Override
+ public boolean hasNext() {
+ return count < numberOfKeys;
+ }
+
+ @Override
+ public StreamRecord<byte[]> next() {
+ ThreadLocalRandom.current().nextBytes(fatArray);
+ changePrefixOfArray(count, fatArray);
+ // make the hashcode of keys different.
+ StreamRecord<byte[]> record =
+ new StreamRecord<>(Arrays.copyOf(fatArray, fatArray.length), 0);
+ count += 1;
+ return record;
+ }
+ };
+ }
+
+ @Override
+ public TypeInformation getTypeInformation() {
+ return PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO;
+ }
+
+ private void changePrefixOfArray(int number, byte[] fatArray) {
+ fatArray[0] = (byte) ((number >> 24) & 0xFF);
+ fatArray[1] = (byte) ((number >> 16) & 0xFF);
+ fatArray[2] = (byte) ((number >> 8) & 0xFF);
+ fatArray[3] = (byte) (number & 0xFF);
+ }
+ }
+
+ protected static class TestKeyedFunction extends KeyedProcessFunction<byte[], byte[], Void> {
+
+ private static final long serialVersionUID = 1L;
+ private final int valueLen = 128;
+
+ private ValueState<byte[]> randomState;
+ private final byte[] stateArray = new byte[valueLen];
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ randomState =
+ this.getRuntimeContext()
+ .getState(new ValueStateDescriptor<>("RandomState", byte[].class));
+ }
+
+ @Override
+ public void processElement(
+ byte[] value,
+ KeyedProcessFunction<byte[], byte[], Void>.Context ctx,
+ Collector<Void> out)
+ throws Exception {
+ ThreadLocalRandom.current().nextBytes(stateArray);
Review comment:
This random cannot guarantee deterministic values.
--
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