Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
doujutun3207
flink
提交
e5ee55bd
F
flink
项目概览
doujutun3207
/
flink
与 Fork 源项目一致
从无法访问的项目Fork
通知
24
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
F
flink
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
e5ee55bd
编写于
10月 16, 2015
作者:
G
gallenvara
提交者:
Fabian Hueske
10月 26, 2015
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[FLINK-2853] Port MutableHashTablePerformanceBenchmark to JMH.
This closes #1267
上级
75a52574
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
137 addition
and
33 deletion
+137
-33
flink-benchmark/pom.xml
flink-benchmark/pom.xml
+9
-2
flink-benchmark/src/test/java/org/apache/flink/benchmark/runtime/operators/hash/MutableHashTablePerformanceBenchmark.java
.../operators/hash/MutableHashTablePerformanceBenchmark.java
+128
-31
未找到文件。
flink-benchmark/pom.xml
浏览文件 @
e5ee55bd
...
...
@@ -56,14 +56,21 @@ under the License.
<artifactId>
jmh-generator-annprocess
</artifactId>
<version>
${jmh.version}
</version>
<scope>
provided
</scope>
</dependency>
<dependency>
<groupId>
org.apache.flink
</groupId>
<artifactId>
flink-runtime
</artifactId>
<version>
0.10-SNAPSHOT
</version>
<type>
test-jar
</type>
<scope>
test
</scope>
</dependency>
<dependency>
<groupId>
org.apache.flink
</groupId>
<artifactId>
flink-
core
</artifactId>
<artifactId>
flink-
java
</artifactId>
<version>
${project.version}
</version>
<type>
test-jar
</type>
<scope>
test
</scope>
</dependency>
</dependency>
</dependencies>
<build>
...
...
flink-
runtime/src/test/java/org/apache/flin
k/runtime/operators/hash/MutableHashTablePerformanceBenchmark.java
→
flink-
benchmark/src/test/java/org/apache/flink/benchmar
k/runtime/operators/hash/MutableHashTablePerformanceBenchmark.java
浏览文件 @
e5ee55bd
...
...
@@ -16,10 +16,11 @@
* limitations under the License.
*/
package
org.apache.flink.runtime.operators.hash
;
package
org.apache.flink.
benchmark.
runtime.operators.hash
;
import
java.io.IOException
;
import
java.util.List
;
import
java.util.concurrent.TimeUnit
;
import
org.apache.flink.api.common.typeutils.TypeComparator
;
import
org.apache.flink.api.common.typeutils.TypePairComparator
;
...
...
@@ -30,6 +31,7 @@ import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
import
org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable
;
import
org.apache.flink.runtime.memory.MemoryAllocationException
;
import
org.apache.flink.runtime.memory.MemoryManager
;
import
org.apache.flink.runtime.operators.hash.MutableHashTable
;
import
org.apache.flink.runtime.operators.testutils.DummyInvokable
;
import
org.apache.flink.runtime.operators.testutils.types.StringPair
;
import
org.apache.flink.runtime.operators.testutils.types.StringPairComparator
;
...
...
@@ -37,13 +39,17 @@ import org.apache.flink.runtime.operators.testutils.types.StringPairPairComparat
import
org.apache.flink.runtime.operators.testutils.types.StringPairSerializer
;
import
org.apache.flink.util.MutableObjectIterator
;
import
org.junit.After
;
import
org.junit.Assert
;
import
org.junit.Before
;
import
org.junit.Test
;
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
;
import
static
org
.
junit
.
Assert
.
fail
;
@State
(
Scope
.
Thread
)
@BenchmarkMode
(
Mode
.
AverageTime
)
@OutputTimeUnit
(
TimeUnit
.
MILLISECONDS
)
public
class
MutableHashTablePerformanceBenchmark
{
private
static
final
AbstractInvokable
MEM_OWNER
=
new
DummyInvokable
();
...
...
@@ -59,8 +65,7 @@ public class MutableHashTablePerformanceBenchmark {
private
static
final
String
COMMENT
=
"this comments should contains a 96 byte data, 100 plus another integer value and seperator char."
;
@Before
@Setup
public
void
setup
()
{
this
.
pairBuildSideAccesssor
=
new
StringPairSerializer
();
this
.
pairProbeSideAccesssor
=
new
StringPairSerializer
();
...
...
@@ -72,7 +77,7 @@ public class MutableHashTablePerformanceBenchmark {
this
.
ioManager
=
new
IOManagerAsync
();
}
@
After
@
TearDown
public
void
tearDown
()
{
// shut down I/O manager and Memory Manager and verify the correct shutdown
this
.
ioManager
.
shutdown
();
...
...
@@ -84,8 +89,8 @@ public class MutableHashTablePerformanceBenchmark {
}
}
@
Test
public
void
compareMutableHashTable
Performance
1
()
throws
IOException
{
@
Benchmark
public
void
compareMutableHashTable
WithBloomFilter
1
()
throws
IOException
{
// ----------------------------------------------90% filtered during probe spill phase-----------------------------------------
// create a build input with 1000000 records with key spread between [0 -- 10000000] with step of 10 for nearby records.
int
buildSize
=
1000000
;
...
...
@@ -98,19 +103,63 @@ public class MutableHashTablePerformanceBenchmark {
int
expectedResult
=
500000
;
long
withBloomFilterCost
=
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
true
);
long
withoutBloomFilterCost
=
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
false
);
this
.
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
true
);
System
.
out
.
println
(
"HybridHashJoin1:"
);
System
.
out
.
println
(
"Build input size: "
+
100
*
buildSize
);
System
.
out
.
println
(
"Probe input size: "
+
100
*
probeSize
);
System
.
out
.
println
(
"Available memory: "
+
this
.
memManager
.
getMemorySize
());
System
.
out
.
println
(
"Probe record be filtered before spill: "
+
(
1
-
(
double
)
probeScope
/
buildScope
)
*
100
+
"% percent."
);
}
@Benchmark
public
void
compareMutableHashTableWithoutBloomFilter1
()
throws
IOException
{
// ----------------------------------------------90% filtered during probe spill phase-----------------------------------------
// create a build input with 1000000 records with key spread between [0 -- 10000000] with step of 10 for nearby records.
int
buildSize
=
1000000
;
int
buildStep
=
10
;
int
buildScope
=
buildStep
*
buildSize
;
// create a probe input with 5000000 records with key spread between [0 -- 1000000] with distance of 1 for nearby records.
int
probeSize
=
5000000
;
int
probeStep
=
1
;
int
probeScope
=
buildSize
;
int
expectedResult
=
500000
;
this
.
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
false
);
System
.
out
.
println
(
"HybridHashJoin1:"
);
System
.
out
.
println
(
"Build input size: "
+
100
*
buildSize
);
System
.
out
.
println
(
"Probe input size: "
+
100
*
probeSize
);
System
.
out
.
println
(
"Available memory: "
+
this
.
memManager
.
getMemorySize
());
System
.
out
.
println
(
"Probe record be filtered before spill: "
+
(
1
-
(
double
)
probeScope
/
buildScope
)
*
100
+
"% percent."
);
}
@Benchmark
public
void
compareMutableHashTableWithBloomFilter2
()
throws
IOException
{
// ----------------------------------------------80% filtered during probe spill phase-----------------------------------------
// create a build input with 1000000 records with key spread between [0 -- 5000000] with step of 5 for nearby records.
int
buildSize
=
1000000
;
int
buildStep
=
5
;
int
buildScope
=
buildStep
*
buildSize
;
// create a probe input with 5000000 records with key spread between [0 -- 1000000] with distance of 1 for nearby records.
int
probeSize
=
5000000
;
int
probeStep
=
1
;
int
probeScope
=
buildSize
;
int
expectedResult
=
1000000
;
this
.
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
true
);
System
.
out
.
println
(
"HybridHashJoin2:"
);
System
.
out
.
println
(
"Build input size: "
+
100
*
buildSize
);
System
.
out
.
println
(
"Probe input size: "
+
100
*
probeSize
);
System
.
out
.
println
(
"Available memory: "
+
this
.
memManager
.
getMemorySize
());
System
.
out
.
println
(
"Probe record be filtered before spill: "
+
(
1
-
(
double
)
probeScope
/
buildScope
)
*
100
+
"% percent."
);
System
.
out
.
println
(
String
.
format
(
"Cost: without bloom filter(%d), with bloom filter(%d)"
,
withoutBloomFilterCost
,
withBloomFilterCost
));
}
@
Test
public
void
compareMutableHashTable
Performance
2
()
throws
IOException
{
@
Benchmark
public
void
compareMutableHashTable
WithoutBloomFilter
2
()
throws
IOException
{
// ----------------------------------------------80% filtered during probe spill phase-----------------------------------------
// create a build input with 1000000 records with key spread between [0 -- 5000000] with step of 5 for nearby records.
int
buildSize
=
1000000
;
...
...
@@ -123,19 +172,40 @@ public class MutableHashTablePerformanceBenchmark {
int
expectedResult
=
1000000
;
long
withBloomFilterCost
=
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
true
);
long
withoutBloomFilterCost
=
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
false
);
this
.
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
false
);
System
.
out
.
println
(
"HybridHashJoin2:"
);
System
.
out
.
println
(
"Build input size: "
+
100
*
buildSize
);
System
.
out
.
println
(
"Probe input size: "
+
100
*
probeSize
);
System
.
out
.
println
(
"Available memory: "
+
this
.
memManager
.
getMemorySize
());
System
.
out
.
println
(
"Probe record be filtered before spill: "
+
(
1
-
(
double
)
probeScope
/
buildScope
)
*
100
+
"% percent."
);
}
@Benchmark
public
void
compareMutableHashTableWithBloomFilter3
()
throws
IOException
{
// ----------------------------------------------50% filtered during probe spill phase-------------------------------------------------
// create a build input with 1000000 records with key spread between [0 -- 2000000] with step of 2 for nearby records.
int
buildSize
=
1000000
;
int
buildStep
=
2
;
int
buildScope
=
buildStep
*
buildSize
;
// create a probe input with 5000000 records with key spread between [0 -- 1000000] with distance of 1 for nearby records.
int
probeSize
=
5000000
;
int
probeStep
=
1
;
int
probeScope
=
buildSize
;
int
expectedResult
=
2500000
;
this
.
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
true
);
System
.
out
.
println
(
"HybridHashJoin3:"
);
System
.
out
.
println
(
"Build input size: "
+
100
*
buildSize
);
System
.
out
.
println
(
"Probe input size: "
+
100
*
probeSize
);
System
.
out
.
println
(
"Available memory: "
+
this
.
memManager
.
getMemorySize
());
System
.
out
.
println
(
"Probe record be filtered before spill: "
+
(
1
-
(
double
)
probeScope
/
buildScope
)
*
100
+
"% percent."
);
System
.
out
.
println
(
String
.
format
(
"Cost: without bloom filter(%d), with bloom filter(%d)"
,
withoutBloomFilterCost
,
withBloomFilterCost
));
}
@
Test
public
void
compareMutableHashTable
Performance
3
()
throws
IOException
{
@
Benchmark
public
void
compareMutableHashTable
WithoutBloomFilter
3
()
throws
IOException
{
// ----------------------------------------------50% filtered during probe spill phase-------------------------------------------------
// create a build input with 1000000 records with key spread between [0 -- 2000000] with step of 2 for nearby records.
int
buildSize
=
1000000
;
...
...
@@ -148,19 +218,40 @@ public class MutableHashTablePerformanceBenchmark {
int
expectedResult
=
2500000
;
long
withoutBloomFilterCost
=
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
false
);
long
withBloomFilterCost
=
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
true
);
this
.
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
false
);
System
.
out
.
println
(
"HybridHashJoin3:"
);
System
.
out
.
println
(
"Build input size: "
+
100
*
buildSize
);
System
.
out
.
println
(
"Probe input size: "
+
100
*
probeSize
);
System
.
out
.
println
(
"Available memory: "
+
this
.
memManager
.
getMemorySize
());
System
.
out
.
println
(
"Probe record be filtered before spill: "
+
(
1
-
(
double
)
probeScope
/
buildScope
)
*
100
+
"% percent."
);
}
@Benchmark
public
void
compareMutableHashTableWithBloomFilter4
()
throws
IOException
{
// ----------------------------------------------0% filtered during probe spill phase-----------------------------------------
// create a build input with 1000000 records with key spread between [0 -- 1000000] with step of 1 for nearby records.
int
buildSize
=
1000000
;
int
buildStep
=
1
;
int
buildScope
=
buildStep
*
buildSize
;
// create a probe input with 5000000 records with key spread between [0 -- 1000000] with distance of 1 for nearby records.
int
probeSize
=
5000000
;
int
probeStep
=
1
;
int
probeScope
=
buildSize
;
int
expectedResult
=
probeSize
/
buildStep
;
this
.
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
true
);
System
.
out
.
println
(
"HybridHashJoin4:"
);
System
.
out
.
println
(
"Build input size: "
+
100
*
buildSize
);
System
.
out
.
println
(
"Probe input size: "
+
100
*
probeSize
);
System
.
out
.
println
(
"Available memory: "
+
this
.
memManager
.
getMemorySize
());
System
.
out
.
println
(
"Probe record be filtered before spill: "
+
(
1
-
(
double
)
probeScope
/
buildScope
)
*
100
+
"% percent."
);
System
.
out
.
println
(
String
.
format
(
"Cost: without bloom filter(%d), with bloom filter(%d)"
,
withoutBloomFilterCost
,
withBloomFilterCost
));
}
@
Test
public
void
compareMutableHashTable
Performance
4
()
throws
IOException
{
@
Benchmark
public
void
compareMutableHashTable
WithoutBloomFilter
4
()
throws
IOException
{
// ----------------------------------------------0% filtered during probe spill phase-----------------------------------------
// create a build input with 1000000 records with key spread between [0 -- 1000000] with step of 1 for nearby records.
int
buildSize
=
1000000
;
...
...
@@ -173,15 +264,13 @@ public class MutableHashTablePerformanceBenchmark {
int
expectedResult
=
probeSize
/
buildStep
;
long
withBloomFilterCost
=
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
true
);
long
withoutBloomFilterCost
=
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
false
);
this
.
hybridHashJoin
(
buildSize
,
buildStep
,
buildScope
,
probeSize
,
probeStep
,
probeScope
,
expectedResult
,
false
);
System
.
out
.
println
(
"HybridHashJoin
5
:"
);
System
.
out
.
println
(
"HybridHashJoin
4
:"
);
System
.
out
.
println
(
"Build input size: "
+
100
*
buildSize
);
System
.
out
.
println
(
"Probe input size: "
+
100
*
probeSize
);
System
.
out
.
println
(
"Available memory: "
+
this
.
memManager
.
getMemorySize
());
System
.
out
.
println
(
"Probe record be filtered before spill: "
+
(
1
-
(
double
)
probeScope
/
buildScope
)
*
100
+
"% percent."
);
System
.
out
.
println
(
String
.
format
(
"Cost: without bloom filter(%d), with bloom filter(%d)"
,
withoutBloomFilterCost
,
withBloomFilterCost
));
}
private
long
hybridHashJoin
(
int
buildSize
,
int
buildStep
,
int
buildScope
,
int
probeSize
,
...
...
@@ -202,7 +291,6 @@ public class MutableHashTablePerformanceBenchmark {
// ----------------------------------------------------------------------------------------
long
start
=
System
.
currentTimeMillis
();
final
MutableHashTable
<
StringPair
,
StringPair
>
join
=
new
MutableHashTable
<
StringPair
,
StringPair
>(
this
.
pairBuildSideAccesssor
,
this
.
pairProbeSideAccesssor
,
this
.
pairBuildSideComparator
,
this
.
pairProbeSideComparator
,
this
.
pairComparator
,
...
...
@@ -221,11 +309,10 @@ public class MutableHashTablePerformanceBenchmark {
Assert
.
assertEquals
(
"Wrong number of records in join result."
,
expectedResultSize
,
numRecordsInJoinResult
);
join
.
close
();
long
cost
=
System
.
currentTimeMillis
()
-
start
;
// ----------------------------------------------------------------------------------------
this
.
memManager
.
release
(
join
.
getFreedMemory
());
return
cost
;
return
1
;
}
...
...
@@ -259,4 +346,14 @@ public class MutableHashTablePerformanceBenchmark {
return
next
(
new
StringPair
());
}
}
public
static
void
main
(
String
[]
args
)
throws
Exception
{
Options
opt
=
new
OptionsBuilder
()
.
include
(
MutableHashTablePerformanceBenchmark
.
class
.
getSimpleName
())
.
warmupIterations
(
2
)
.
measurementIterations
(
2
)
.
forks
(
1
)
.
build
();
new
Runner
(
opt
).
run
();
}
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录