This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-benchmarks.git
The following commit(s) were added to refs/heads/master by this push:
new 517f1e7 [FLINK-24300] Added benchmarks for idling chained sources
517f1e7 is described below
commit 517f1e7301b6a7086a3d8b70067da88ba6294022
Author: Dawid Wysakowicz <dwysakow...@apache.org>
AuthorDate: Thu Sep 16 20:16:20 2021 +0200
[FLINK-24300] Added benchmarks for idling chained sources
This closes #31
---
pom.xml | 6 ++
.../flink/benchmark/MultipleInputBenchmark.java | 106 +++++++++++++++++++--
2 files changed, 105 insertions(+), 7 deletions(-)
diff --git a/pom.xml b/pom.xml
index f0f0bda..f92f166 100644
--- a/pom.xml
+++ b/pom.xml
@@ -201,6 +201,12 @@ under the License.
<type>jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${flink.version}</version>
+ <type>test-jar</type>
+ </dependency>
<!-- serialization benchmark requirements -->
diff --git
a/src/main/java/org/apache/flink/benchmark/MultipleInputBenchmark.java
b/src/main/java/org/apache/flink/benchmark/MultipleInputBenchmark.java
index f070c59..53c6115 100644
--- a/src/main/java/org/apache/flink/benchmark/MultipleInputBenchmark.java
+++ b/src/main/java/org/apache/flink/benchmark/MultipleInputBenchmark.java
@@ -18,14 +18,27 @@
package org.apache.flink.benchmark;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.lib.NumberSequenceSource;
+import org.apache.flink.api.connector.source.mocks.MockSource;
+import org.apache.flink.api.connector.source.mocks.MockSourceReader;
+import org.apache.flink.api.connector.source.mocks.MockSourceSplit;
import org.apache.flink.benchmark.functions.LongSource;
import org.apache.flink.benchmark.functions.QueuingLongSource;
import org.apache.flink.benchmark.operators.MultiplyByTwoOperatorFactory;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.MultipleConnectedStreams;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
import
org.apache.flink.streaming.api.transformations.MultipleInputTransformation;
import org.openjdk.jmh.annotations.Benchmark;
@@ -36,18 +49,19 @@ import org.openjdk.jmh.runner.options.Options;
import org.openjdk.jmh.runner.options.OptionsBuilder;
import org.openjdk.jmh.runner.options.VerboseMode;
-public class MultipleInputBenchmark extends BenchmarkBase {
+import java.util.concurrent.CompletableFuture;
+public class MultipleInputBenchmark extends BenchmarkBase {
public static final int RECORDS_PER_INVOCATION =
TwoInputBenchmark.RECORDS_PER_INVOCATION;
public static final int ONE_IDLE_RECORDS_PER_INVOCATION =
TwoInputBenchmark.ONE_IDLE_RECORDS_PER_INVOCATION;
public static final long CHECKPOINT_INTERVAL_MS =
TwoInputBenchmark.CHECKPOINT_INTERVAL_MS;
public static void main(String[] args)
- throws RunnerException {
+ throws RunnerException {
Options options = new OptionsBuilder()
- .verbosity(VerboseMode.NORMAL)
- .include(".*" +
MultipleInputBenchmark.class.getSimpleName() + ".*")
- .build();
+ .verbosity(VerboseMode.NORMAL)
+ .include(".*" +
MultipleInputBenchmark.class.getSimpleName() + ".*")
+ .build();
new Runner(options).run();
}
@@ -82,10 +96,88 @@ public class MultipleInputBenchmark extends BenchmarkBase {
env.execute();
}
+ @Benchmark
+ @OperationsPerInvocation(RECORDS_PER_INVOCATION)
+ public void multiInputChainedIdleSource(FlinkEnvironmentContext
context) throws Exception {
+ final StreamExecutionEnvironment env = context.env;
+ env.getConfig().enableObjectReuse();
+
+ final DataStream<Long> source1 =
+ env.fromSource(
+ new NumberSequenceSource(1L,
RECORDS_PER_INVOCATION),
+
WatermarkStrategy.noWatermarks(),
+ "source-1");
+
+ final DataStreamSource<Integer> source2 =
+ env.fromSource(new IdlingSource(1),
WatermarkStrategy.noWatermarks(), "source-2");
+
+ MultipleInputTransformation<Long> transform = new
MultipleInputTransformation<>(
+ "custom operator",
+ new MultiplyByTwoOperatorFactory(),
+ BasicTypeInfo.LONG_TYPE_INFO,
+ 1);
+
+ transform.addInput(((DataStream<?>)
source1).getTransformation());
+ transform.addInput(((DataStream<?>)
source2).getTransformation());
+
transform.setChainingStrategy(ChainingStrategy.HEAD_WITH_SOURCES);
+
+ env.addOperator(transform);
+ new
MultipleConnectedStreams(env).transform(transform).addSink(new
SinkClosingIdlingSource()).setParallelism(1);
+ context.execute();
+ }
+
+ private static class IdlingSource extends MockSource {
+ private static CompletableFuture<Void> canFinish = new
CompletableFuture<>();
+
+ public static void signalCanFinish() {
+ canFinish.complete(null);
+ }
+
+ public static void reset() {
+ canFinish.completeExceptionally(new
IllegalStateException("State has been reset"));
+ canFinish = new CompletableFuture<>();
+ }
+
+ public IdlingSource(int numSplits) {
+ super(Boundedness.BOUNDED, numSplits, true, true);
+ }
+
+ @Override
+ public SourceReader<Integer, MockSourceSplit> createReader(
+ SourceReaderContext readerContext) {
+ return new MockSourceReader(true, true) {
+ @Override
+ public InputStatus
pollNext(ReaderOutput<Integer> sourceOutput) {
+ if (canFinish.isDone() &&
!canFinish.isCompletedExceptionally()) {
+ return InputStatus.END_OF_INPUT;
+ } else {
+ return
InputStatus.NOTHING_AVAILABLE;
+ }
+ }
+
+ @Override
+ public synchronized CompletableFuture<Void>
isAvailable() {
+ return canFinish;
+ }
+ };
+ }
+ }
+
+ private static class SinkClosingIdlingSource implements
SinkFunction<Long> {
+ private int recordsSoFar = 0;
+
+ @Override
+ public void invoke(Long value) {
+ if (++recordsSoFar >= RECORDS_PER_INVOCATION) {
+ IdlingSource.signalCanFinish();
+ }
+ }
+ }
+
private static void connectAndDiscard(
StreamExecutionEnvironment env,
- DataStreamSource<Long> source1,
- DataStreamSource<Long> source2) {
+ DataStream<?> source1,
+ DataStream<?> source2) {
MultipleInputTransformation<Long> transform = new
MultipleInputTransformation<>(
"custom operator",
new MultiplyByTwoOperatorFactory(),
