Skip to content

Commit

Permalink
[FLINK-2920] [tests] Port KryoVersusAvroMinibenchmark to JMH.
Browse files Browse the repository at this point in the history 
This closes apache#1302
  • Loading branch information
@gallenvara @fhueske
gallenvara authored and fhueske committed Oct 29, 2015
1 parent 3abbcd1 commit a8f1be9
Show file tree
Hide file tree
Showing 2 changed files with 154 additions and 159 deletions.
173 changes: 86 additions & 87 deletions flink-benchmark/pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,99 +16,98 @@ software distributed under the License is distributed on an
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

<modelVersion>4.0.0</modelVersion>
<modelVersion>4.0.0</modelVersion>

<parent>
<groupId>org.apache.flink</groupId>
<artifactId>flink-parent</artifactId>
<version>1.0-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>
<parent>
<groupId>org.apache.flink</groupId>
<artifactId>flink-parent</artifactId>
<version>1.0-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>

<artifactId>flink-benchmark</artifactId>
<name>flink-benchmark</name>
<packaging>jar</packaging>
<artifactId>flink-benchmark</artifactId>
<name>flink-benchmark</name>
<packaging>jar</packaging>

<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<jmh.version>1.11</jmh.version>
<uberjar.name>benchmarks</uberjar.name>
</properties>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<jmh.version>1.11</jmh.version>
<uberjar.name>benchmarks</uberjar.name>
</properties>

<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.openjdk.jmh</groupId>
<artifactId>jmh-core</artifactId>
<version>${jmh.version}</version>
</dependency>
<dependency>
<groupId>org.openjdk.jmh</groupId>
<artifactId>jmh-generator-annprocess</artifactId>
<version>${jmh.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime</artifactId>
<version>1.0-SNAPSHOT</version>
<type>test-jar</type>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${project.version}</version>
<type>test-jar</type>
<scope>test</scope>
</dependency>
</dependencies>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.openjdk.jmh</groupId>
<artifactId>jmh-core</artifactId>
<version>${jmh.version}</version>
</dependency>
<dependency>
<groupId>org.openjdk.jmh</groupId>
<artifactId>jmh-generator-annprocess</artifactId>
<version>${jmh.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime</artifactId>
<version>${project.version}</version>
<type>test-jar</type>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${project.version}</version>
<type>test-jar</type>
<scope>test</scope>
</dependency>
</dependencies>

<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<finalName>${uberjar.name}</finalName>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>org.openjdk.jmh.Main</mainClass>
</transformer>
</transformers>
<filters>
<filter>
<!--
Shading signed JARs will fail without this.
http://stackoverflow.com/questions/999489/invalid-signature-file-when-attempting-to-run-a-jar
-->
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<finalName>${uberjar.name}</finalName>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>org.openjdk.jmh.Main</mainClass>
</transformer>
</transformers>
<filters>
<filter>
<!--
Shading signed JARs will fail without this.
http://stackoverflow.com/questions/999489/invalid-signature-file-when-attempting-to-run-a-jar
-->
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
140 changes: 68 additions & 72 deletions ...ime/kryo/KryoVersusAvroMinibenchmark.java → ...ime/kryo/KryoVersusAvroMinibenchmark.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
* limitations under the License.
*/

package org.apache.flink.api.java.typeutils.runtime.kryo;
package org.apache.flink.benchmark.api.java.typeutils.runtime.kryo;

import java.io.EOFException;
import java.io.IOException;
Expand All @@ -26,6 +26,7 @@
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
Expand All @@ -36,88 +37,83 @@
import org.apache.flink.api.java.typeutils.runtime.TestDataOutputSerializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.MemoryUtils;

import org.openjdk.jmh.annotations.*;
import org.openjdk.jmh.runner.Runner;
import org.openjdk.jmh.runner.options.Options;
import org.openjdk.jmh.runner.options.OptionsBuilder;

@State(Scope.Thread)
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
public class KryoVersusAvroMinibenchmark {

private static final long SEED = 94762389741692387L;

private static final Random rnd = new Random(SEED);

private static final int NUM_ELEMENTS = 100000;

private static final int NUM_RUNS = 10;



public static void main(String[] args) throws Exception {

final MyType[] elements = new MyType[NUM_ELEMENTS];

@Param({"1","2","3","4","5","6","7","8","9","10"})
private static int runTime;

private MyType[] elements ;

private MyType dummy;

@Setup
public void init() {
this.elements = new MyType[NUM_ELEMENTS];
for (int i = 0; i < NUM_ELEMENTS; i++) {
elements[i] = MyType.getRandom();
this.elements[i] = MyType.getRandom();
}

final MyType dummy = new MyType();

long[] timesAvro = new long[NUM_RUNS];
long[] timesKryo = new long[NUM_RUNS];

for (int i = 0; i < NUM_RUNS; i++) {
System.out.println("----------------- Starting run " + i + " ---------------------");

System.out.println("Avro serializer");
{
final TestDataOutputSerializer outView = new TestDataOutputSerializer(100000000);
final AvroSerializer<MyType> serializer = new AvroSerializer<MyType>(MyType.class);

long start = System.nanoTime();

for (int k = 0; k < NUM_ELEMENTS; k++) {
serializer.serialize(elements[k], outView);
}

final DataInputDeserializer inView = new DataInputDeserializer(outView.wrapAsByteBuffer());
for (int k = 0; k < NUM_ELEMENTS; k++) {
serializer.deserialize(dummy, inView);
}

long elapsed = System.nanoTime() - start;
System.out.println("Took: " + (elapsed / 1000000) + " msecs");
timesAvro[i] = elapsed;
}

System.gc();

System.out.println("Kryo serializer");
{
final TestDataOutputSerializer outView = new TestDataOutputSerializer(100000000);
ExecutionConfig conf = new ExecutionConfig();
conf.registerKryoType(MyType.class);
conf.enableForceKryo();
TypeInformation<MyType> typeInfo = new GenericTypeInfo<MyType>(MyType.class);
final TypeSerializer<MyType> serializer = typeInfo.createSerializer(conf);

long start = System.nanoTime();

for (int k = 0; k < NUM_ELEMENTS; k++) {
serializer.serialize(elements[k], outView);
}

final DataInputDeserializer inView = new DataInputDeserializer(outView.wrapAsByteBuffer());
for (int k = 0; k < NUM_ELEMENTS; k++) {
serializer.deserialize(dummy, inView);
}

long elapsed = System.nanoTime() - start;
System.out.println("Took: " + (elapsed / 1000000) + " msecs");
timesKryo[i] = elapsed;
}
this.dummy = new MyType();
}

@Benchmark
public void avroSerializer() throws IOException {
final TestDataOutputSerializer outView = new TestDataOutputSerializer(10000000);
final AvroSerializer<MyType> serializer = new AvroSerializer<MyType>(MyType.class);

for (int k = 0; k < NUM_ELEMENTS; k++) {
serializer.serialize(elements[k], outView);
}

final DataInputDeserializer inView = new DataInputDeserializer(outView.wrapAsByteBuffer());
for (int k = 0; k < NUM_ELEMENTS; k++) {
serializer.deserialize(dummy, inView);
}
}






@Benchmark
public void kryoSerializer() throws IOException {
final TestDataOutputSerializer outView = new TestDataOutputSerializer(10000000);
ExecutionConfig conf = new ExecutionConfig();
conf.registerKryoType(MyType.class);
conf.enableForceKryo();
TypeInformation<MyType> typeInfo = new GenericTypeInfo<MyType>(MyType.class);
final TypeSerializer<MyType> serializer = typeInfo.createSerializer(conf);

for (int k = 0; k < NUM_ELEMENTS; k++) {
serializer.serialize(elements[k], outView);
}

final DataInputDeserializer inView = new DataInputDeserializer(outView.wrapAsByteBuffer());
for (int k = 0; k < NUM_ELEMENTS; k++) {
serializer.deserialize(dummy, inView);
}
}

public static void main(String[] args) throws Exception {

Options opt = new OptionsBuilder()
.include(KryoVersusAvroMinibenchmark.class.getSimpleName())
.warmupIterations(2)
.measurementIterations(2)
.forks(1)
.build();
new Runner(opt).run();
}

public static class MyType {

private String theString;
Expand Down

0 comments on commit a8f1be9

Please sign in to comment.