You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by nk...@apache.org on 2018/07/19 19:08:52 UTC
[2/2] flink git commit: [FLINK-9435][java] optimise
ComparableKeySelector and ArrayKeySelector for more efficient Tuple creation
[FLINK-9435][java] optimise ComparableKeySelector and ArrayKeySelector for more efficient Tuple creation
Benchmark results (2 runs) by running the benchmarks from
https://github.com/dataArtisans/flink-benchmarks/pull/5:
Benchmark Mode Cnt Score Error Units
================= old =================
KeyByBenchmarks.arrayKeyBy thrpt 30 1151.305 ± 21.096 ops/ms
KeyByBenchmarks.arrayKeyBy thrpt 30 1117.486 ± 43.508 ops/ms
KeyByBenchmarks.tupleKeyBy thrpt 30 1659.634 ± 28.627 ops/ms
KeyByBenchmarks.tupleKeyBy thrpt 30 1554.265 ± 82.604 ops/ms
================= new =================
KeyByBenchmarks.arrayKeyBy thrpt 30 1150.552 ± 51.185 ops/ms
KeyByBenchmarks.arrayKeyBy thrpt 30 1195.777 ± 10.621 ops/ms
KeyByBenchmarks.tupleKeyBy thrpt 30 1743.633 ± 27.109 ops/ms
KeyByBenchmarks.tupleKeyBy thrpt 30 1697.885 ± 22.101 ops/ms
This closes #6115.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/95eadfe1
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/95eadfe1
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/95eadfe1
Branch: refs/heads/master
Commit: 95eadfe15203ee0ab1459a9ade943234d9d6e7ce
Parents: 5857f55
Author: Nico Kruber <ni...@data-artisans.com>
Authored: Fri May 25 00:09:37 2018 +0200
Committer: Nico Kruber <ni...@data-artisans.com>
Committed: Thu Jul 19 17:07:09 2018 +0200
----------------------------------------------------------------------
.../org/apache/flink/api/java/tuple/Tuple.java | 32 ++++++++++++++++++++
.../flink/api/java/tuple/TupleGenerator.java | 11 +++++++
.../streaming/util/keys/KeySelectorUtil.java | 10 +++---
3 files changed, 47 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/95eadfe1/flink-core/src/main/java/org/apache/flink/api/java/tuple/Tuple.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/tuple/Tuple.java b/flink-core/src/main/java/org/apache/flink/api/java/tuple/Tuple.java
index c282c59..7ce38f8 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/tuple/Tuple.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/tuple/Tuple.java
@@ -113,6 +113,38 @@ public abstract class Tuple implements java.io.Serializable {
// BEGIN_OF_TUPLE_DEPENDENT_CODE
// GENERATED FROM org.apache.flink.api.java.tuple.TupleGenerator.
+ public static Tuple newInstance(int arity) {
+ switch (arity) {
+ case 0: return Tuple0.INSTANCE;
+ case 1: return new Tuple1();
+ case 2: return new Tuple2();
+ case 3: return new Tuple3();
+ case 4: return new Tuple4();
+ case 5: return new Tuple5();
+ case 6: return new Tuple6();
+ case 7: return new Tuple7();
+ case 8: return new Tuple8();
+ case 9: return new Tuple9();
+ case 10: return new Tuple10();
+ case 11: return new Tuple11();
+ case 12: return new Tuple12();
+ case 13: return new Tuple13();
+ case 14: return new Tuple14();
+ case 15: return new Tuple15();
+ case 16: return new Tuple16();
+ case 17: return new Tuple17();
+ case 18: return new Tuple18();
+ case 19: return new Tuple19();
+ case 20: return new Tuple20();
+ case 21: return new Tuple21();
+ case 22: return new Tuple22();
+ case 23: return new Tuple23();
+ case 24: return new Tuple24();
+ case 25: return new Tuple25();
+ default: throw new IllegalArgumentException("The tuple arity must be in [0, " + MAX_ARITY + "].");
+ }
+ }
+
private static final Class<?>[] CLASSES = new Class<?>[] {
Tuple0.class, Tuple1.class, Tuple2.class, Tuple3.class, Tuple4.class, Tuple5.class, Tuple6.class, Tuple7.class, Tuple8.class, Tuple9.class, Tuple10.class, Tuple11.class, Tuple12.class, Tuple13.class, Tuple14.class, Tuple15.class, Tuple16.class, Tuple17.class, Tuple18.class, Tuple19.class, Tuple20.class, Tuple21.class, Tuple22.class, Tuple23.class, Tuple24.class, Tuple25.class
};
http://git-wip-us.apache.org/repos/asf/flink/blob/95eadfe1/flink-core/src/test/java/org/apache/flink/api/java/tuple/TupleGenerator.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/tuple/TupleGenerator.java b/flink-core/src/test/java/org/apache/flink/api/java/tuple/TupleGenerator.java
index bd5598a..d684967 100644
--- a/flink-core/src/test/java/org/apache/flink/api/java/tuple/TupleGenerator.java
+++ b/flink-core/src/test/java/org/apache/flink/api/java/tuple/TupleGenerator.java
@@ -133,6 +133,17 @@ class TupleGenerator {
private static void modifyTupleType(File root) throws IOException {
// generate code
StringBuilder sb = new StringBuilder();
+ sb.append("\tpublic static Tuple newInstance(int arity) {\n");
+ sb.append("\t\tswitch (arity) {\n");
+ // special case for Tuple0:
+ sb.append("\t\t\tcase 0: return Tuple0.INSTANCE;\n");
+ for (int i = FIRST; i <= LAST; i++) {
+ sb.append("\t\t\tcase ").append(i).append(": return new Tuple").append(i).append("();\n");
+ }
+ sb.append("\t\t\tdefault: throw new IllegalArgumentException(\"The tuple arity must be in [0, \" + MAX_ARITY + \"].\");\n");
+ sb.append("\t\t}\n");
+ sb.append("\t}\n\n");
+
sb.append("\tprivate static final Class<?>[] CLASSES = new Class<?>[] {\n\t\tTuple0.class");
for (int i = FIRST; i <= LAST; i++) {
sb.append(", Tuple").append(i).append(".class");
http://git-wip-us.apache.org/repos/asf/flink/blob/95eadfe1/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/keys/KeySelectorUtil.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/keys/KeySelectorUtil.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/keys/KeySelectorUtil.java
index 27ce573..ab608ef 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/keys/KeySelectorUtil.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/keys/KeySelectorUtil.java
@@ -180,8 +180,8 @@ public final class KeySelectorUtil {
}
@Override
- public Tuple getKey(IN value) throws Exception {
- Tuple key = Tuple.getTupleClass(keyLength).newInstance();
+ public Tuple getKey(IN value) {
+ Tuple key = Tuple.newInstance(keyLength);
comparator.extractKeys(value, keyArray, 0);
for (int i = 0; i < keyLength; i++) {
key.setField(keyArray[i], i);
@@ -210,18 +210,16 @@ public final class KeySelectorUtil {
private static final long serialVersionUID = 1L;
private final int[] fields;
- private final Class<? extends Tuple> tupleClass;
private transient TupleTypeInfo<Tuple> returnType;
ArrayKeySelector(int[] fields, TupleTypeInfo<Tuple> returnType) {
this.fields = requireNonNull(fields);
this.returnType = requireNonNull(returnType);
- this.tupleClass = Tuple.getTupleClass(fields.length);
}
@Override
- public Tuple getKey(IN value) throws Exception {
- Tuple key = tupleClass.newInstance();
+ public Tuple getKey(IN value) {
+ Tuple key = Tuple.newInstance(fields.length);
for (int i = 0; i < fields.length; i++) {
key.setField(Array.get(value, fields[i]), i);
}