Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
doujutun3207
flink
提交
7330dc70
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,发现更多精彩内容 >>
提交
7330dc70
编写于
11月 19, 2013
作者:
A
Aljoscha Krettek
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
Move TPCH3 IT case from TestBase to TestBase2
上级
bed28cb2
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
16 addition
and
75 deletion
+16
-75
pact/pact-tests/src/test/java/eu/stratosphere/pact/test/pactPrograms/TPCHQuery3ITCase.java
...stratosphere/pact/test/pactPrograms/TPCHQuery3ITCase.java
+11
-64
pact/pact-tests/src/test/java/eu/stratosphere/pact/test/scalaPactPrograms/TPCHQuery3ITCase.java
...osphere/pact/test/scalaPactPrograms/TPCHQuery3ITCase.java
+5
-11
未找到文件。
pact/pact-tests/src/test/java/eu/stratosphere/pact/test/pactPrograms/TPCHQuery3ITCase.java
浏览文件 @
7330dc70
...
...
@@ -32,10 +32,10 @@ import eu.stratosphere.pact.compiler.PactCompiler;
import
eu.stratosphere.pact.compiler.plan.candidate.OptimizedPlan
;
import
eu.stratosphere.pact.compiler.plantranslate.NepheleJobGraphGenerator
;
import
eu.stratosphere.pact.example.relational.TPCHQuery3
;
import
eu.stratosphere.pact.test.util.TestBase
;
import
eu.stratosphere.pact.test.util.TestBase
2
;
@RunWith
(
Parameterized
.
class
)
public
class
TPCHQuery3ITCase
extends
TestBase
{
public
class
TPCHQuery3ITCase
extends
TestBase
2
{
private
static
final
Log
LOG
=
LogFactory
.
getLog
(
TPCHQuery3ITCase
.
class
);
...
...
@@ -138,84 +138,31 @@ public class TPCHQuery3ITCase extends TestBase {
@Override
protected
void
preSubmit
()
throws
Exception
{
ordersPath
=
getFilesystemProvider
().
getTempDirPath
()
+
"/orders"
;
lineitemsPath
=
getFilesystemProvider
().
getTempDirPath
()
+
"/lineitems"
;
resultPath
=
getFilesystemProvider
().
getTempDirPath
()
+
"/result"
;
String
[]
splits
=
splitInputString
(
ORDERS
,
'\n'
,
4
);
getFilesystemProvider
().
createDir
(
ordersPath
);
for
(
int
i
=
0
;
i
<
splits
.
length
;
i
++)
{
getFilesystemProvider
().
createFile
(
ordersPath
+
"/part_"
+
i
+
".txt"
,
splits
[
i
]);
if
(
LOG
.
isDebugEnabled
())
LOG
.
debug
(
"Orders Part "
+
(
i
+
1
)
+
":\n>"
+
splits
[
i
]
+
"<"
);
}
splits
=
splitInputString
(
LINEITEMS
,
'\n'
,
4
);
getFilesystemProvider
().
createDir
(
lineitemsPath
);
for
(
int
i
=
0
;
i
<
splits
.
length
;
i
++)
{
getFilesystemProvider
().
createFile
(
lineitemsPath
+
"/part_"
+
i
+
".txt"
,
splits
[
i
]);
if
(
LOG
.
isDebugEnabled
())
LOG
.
debug
(
"Lineitems Part "
+
(
i
+
1
)
+
":\n>"
+
splits
[
i
]
+
"<"
);
}
ordersPath
=
createTempFile
(
"orders"
,
ORDERS
);
lineitemsPath
=
createTempFile
(
"lineitems"
,
LINEITEMS
);
resultPath
=
getTempDirPath
(
"result"
);
}
@Override
protected
JobGraph
getJobGraph
()
throws
Exception
{
protected
Plan
getPactPlan
()
{
TPCHQuery3
tpch3
=
new
TPCHQuery3
();
Plan
plan
=
tpch3
.
getPlan
(
return
tpch3
.
getPlan
(
config
.
getString
(
"TPCHQuery3Test#NoSubtasks"
,
"1"
),
getFilesystemProvider
().
getURIPrefix
()+
ordersPath
,
getFilesystemProvider
().
getURIPrefix
()+
lineitemsPath
,
getFilesystemProvider
().
getURIPrefix
()+
resultPath
);
PactCompiler
pc
=
new
PactCompiler
(
new
DataStatistics
());
OptimizedPlan
op
=
pc
.
compile
(
plan
);
NepheleJobGraphGenerator
jgg
=
new
NepheleJobGraphGenerator
();
return
jgg
.
compileJobGraph
(
op
);
ordersPath
,
lineitemsPath
,
resultPath
);
}
@Override
protected
void
postSubmit
()
throws
Exception
{
compareResultsByLinesInMemory
(
EXPECTED_RESULT
,
resultPath
);
}
@Override
public
void
stopCluster
()
throws
Exception
{
getFilesystemProvider
().
delete
(
ordersPath
,
true
);
getFilesystemProvider
().
delete
(
lineitemsPath
,
true
);
getFilesystemProvider
().
delete
(
resultPath
,
true
);
super
.
stopCluster
();
}
@Parameters
public
static
Collection
<
Object
[]>
getConfigurations
()
{
ArrayList
<
Configuration
>
tConfigs
=
new
ArrayList
<
Configuration
>();
Configuration
config
=
new
Configuration
();
config
.
setInteger
(
"TPCHQuery3Test#NoSubtasks"
,
4
);
tConfigs
.
add
(
config
);
return
toParameterList
(
tConfigs
);
}
private
String
[]
splitInputString
(
String
inputString
,
char
splitChar
,
int
noSplits
)
{
String
splitString
=
inputString
.
toString
();
String
[]
splits
=
new
String
[
noSplits
];
int
partitionSize
=
(
splitString
.
length
()
/
noSplits
)
-
2
;
// split data file and copy parts
for
(
int
i
=
0
;
i
<
noSplits
-
1
;
i
++)
{
int
cutPos
=
splitString
.
indexOf
(
splitChar
,
(
partitionSize
<
splitString
.
length
()
?
partitionSize
:
(
splitString
.
length
()
-
1
)));
splits
[
i
]
=
splitString
.
substring
(
0
,
cutPos
)
+
"\n"
;
splitString
=
splitString
.
substring
(
cutPos
+
1
);
}
splits
[
noSplits
-
1
]
=
splitString
;
return
splits
;
return
toParameterList
(
config
);
}
}
pact/pact-tests/src/test/java/eu/stratosphere/pact/test/scalaPactPrograms/TPCHQuery3ITCase.java
浏览文件 @
7330dc70
...
...
@@ -35,20 +35,14 @@ public class TPCHQuery3ITCase extends eu.stratosphere.pact.test.pactPrograms.TPC
}
@Override
protected
JobGraph
getJobGraph
()
throws
Exception
{
protected
Plan
getPactPlan
()
{
TPCHQuery3
tpch3
=
new
TPCHQuery3
();
Plan
plan
=
tpch3
.
getScalaPlan
(
return
tpch3
.
getScalaPlan
(
config
.
getInteger
(
"TPCHQuery3Test#NoSubtasks"
,
1
),
getFilesystemProvider
().
getURIPrefix
()+
ordersPath
,
getFilesystemProvider
().
getURIPrefix
()+
lineitemsPath
,
getFilesystemProvider
().
getURIPrefix
()+
resultPath
,
ordersPath
,
lineitemsPath
,
resultPath
,
'F'
,
1993
,
"5"
);
PactCompiler
pc
=
new
PactCompiler
(
new
DataStatistics
());
OptimizedPlan
op
=
pc
.
compile
(
plan
);
NepheleJobGraphGenerator
jgg
=
new
NepheleJobGraphGenerator
();
return
jgg
.
compileJobGraph
(
op
);
}
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录