Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
doujutun3207
flink
提交
65bf092d
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,发现更多精彩内容 >>
提交
65bf092d
编写于
10月 03, 2014
作者:
S
Stephan Ewen
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[FLINK-1110] Adjust collection based runtime and tests for classloaders in runtime context
上级
ff5ddd50
变更
9
显示空白变更内容
内联
并排
Showing
9 changed file
with
24 addition
and
20 deletion
+24
-20
flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
...apache/flink/api/common/operators/CollectionExecutor.java
+10
-6
flink-core/src/test/java/org/apache/flink/api/common/operators/base/FlatMapOperatorCollectionTest.java
.../common/operators/base/FlatMapOperatorCollectionTest.java
+1
-1
flink-core/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
...flink/api/common/operators/base/JoinOperatorBaseTest.java
+2
-2
flink-core/src/test/java/org/apache/flink/api/common/operators/base/MapOperatorTest.java
...ache/flink/api/common/operators/base/MapOperatorTest.java
+2
-2
flink-core/src/test/java/org/apache/flink/api/common/operators/base/PartitionMapOperatorTest.java
...k/api/common/operators/base/PartitionMapOperatorTest.java
+2
-2
flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java
.../common/operators/base/CoGroupOperatorCollectionTest.java
+1
-1
flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
...nk/api/common/operators/base/GroupReduceOperatorTest.java
+2
-2
flink-java/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
...flink/api/common/operators/base/JoinOperatorBaseTest.java
+2
-2
flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
...e/flink/api/common/operators/base/ReduceOperatorTest.java
+2
-2
未找到文件。
flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
浏览文件 @
65bf092d
...
...
@@ -66,6 +66,8 @@ public class CollectionExecutor {
private
final
Map
<
String
,
Aggregator
<?>>
aggregators
;
private
final
ClassLoader
classLoader
;
private
final
boolean
mutableObjectSafeMode
;
// --------------------------------------------------------------------------------------------
...
...
@@ -81,6 +83,8 @@ public class CollectionExecutor {
this
.
accumulators
=
new
HashMap
<
String
,
Accumulator
<?,?>>();
this
.
previousAggregates
=
new
HashMap
<
String
,
Value
>();
this
.
aggregators
=
new
HashMap
<
String
,
Aggregator
<?>>();
this
.
classLoader
=
getClass
().
getClassLoader
();
}
// --------------------------------------------------------------------------------------------
...
...
@@ -181,8 +185,8 @@ public class CollectionExecutor {
// build the runtime context and compute broadcast variables, if necessary
RuntimeUDFContext
ctx
;
if
(
RichFunction
.
class
.
isAssignableFrom
(
typedOp
.
getUserCodeWrapper
().
getUserCodeClass
()))
{
ctx
=
superStep
==
0
?
new
RuntimeUDFContext
(
operator
.
getName
(),
1
,
0
)
:
new
IterationRuntimeUDFContext
(
operator
.
getName
(),
1
,
0
,
superStep
);
ctx
=
superStep
==
0
?
new
RuntimeUDFContext
(
operator
.
getName
(),
1
,
0
,
getClass
().
getClassLoader
()
)
:
new
IterationRuntimeUDFContext
(
operator
.
getName
(),
1
,
0
,
superStep
,
classLoader
);
for
(
Map
.
Entry
<
String
,
Operator
<?>>
bcInputs
:
operator
.
getBroadcastInputs
().
entrySet
())
{
List
<?>
bcData
=
execute
(
bcInputs
.
getValue
());
...
...
@@ -223,8 +227,8 @@ public class CollectionExecutor {
// build the runtime context and compute broadcast variables, if necessary
RuntimeUDFContext
ctx
;
if
(
RichFunction
.
class
.
isAssignableFrom
(
typedOp
.
getUserCodeWrapper
().
getUserCodeClass
()))
{
ctx
=
superStep
==
0
?
new
RuntimeUDFContext
(
operator
.
getName
(),
1
,
0
)
:
new
IterationRuntimeUDFContext
(
operator
.
getName
(),
1
,
0
,
superStep
);
ctx
=
superStep
==
0
?
new
RuntimeUDFContext
(
operator
.
getName
(),
1
,
0
,
classLoader
)
:
new
IterationRuntimeUDFContext
(
operator
.
getName
(),
1
,
0
,
superStep
,
classLoader
);
for
(
Map
.
Entry
<
String
,
Operator
<?>>
bcInputs
:
operator
.
getBroadcastInputs
().
entrySet
())
{
List
<?>
bcData
=
execute
(
bcInputs
.
getValue
());
...
...
@@ -478,8 +482,8 @@ public class CollectionExecutor {
private
final
int
superstep
;
public
IterationRuntimeUDFContext
(
String
name
,
int
numParallelSubtasks
,
int
subtaskIndex
,
int
superstep
)
{
super
(
name
,
numParallelSubtasks
,
subtaskIndex
);
public
IterationRuntimeUDFContext
(
String
name
,
int
numParallelSubtasks
,
int
subtaskIndex
,
int
superstep
,
ClassLoader
classloader
)
{
super
(
name
,
numParallelSubtasks
,
subtaskIndex
,
classloader
);
this
.
superstep
=
superstep
;
}
...
...
flink-core/src/test/java/org/apache/flink/api/common/operators/base/FlatMapOperatorCollectionTest.java
浏览文件 @
65bf092d
...
...
@@ -65,7 +65,7 @@ public class FlatMapOperatorCollectionTest implements Serializable {
private
void
testExecuteOnCollection
(
FlatMapFunction
<
String
,
String
>
udf
,
List
<
String
>
input
,
boolean
mutableSafe
)
throws
Exception
{
// run on collections
final
List
<
String
>
result
=
getTestFlatMapOperator
(
udf
)
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
"Test UDF"
,
4
,
0
),
mutableSafe
);
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
"Test UDF"
,
4
,
0
,
null
),
mutableSafe
);
Assert
.
assertEquals
(
input
.
size
(),
result
.
size
());
Assert
.
assertEquals
(
input
,
result
);
...
...
flink-core/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
浏览文件 @
65bf092d
...
...
@@ -110,8 +110,8 @@ public class JoinOperatorBaseTest implements Serializable {
try
{
List
<
Integer
>
resultSafe
=
base
.
executeOnCollections
(
inputData1
,
inputData2
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
),
true
);
List
<
Integer
>
resultRegular
=
base
.
executeOnCollections
(
inputData1
,
inputData2
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
),
false
);
List
<
Integer
>
resultSafe
=
base
.
executeOnCollections
(
inputData1
,
inputData2
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
,
null
),
true
);
List
<
Integer
>
resultRegular
=
base
.
executeOnCollections
(
inputData1
,
inputData2
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
,
null
),
false
);
assertEquals
(
expected
,
resultSafe
);
assertEquals
(
expected
,
resultRegular
);
...
...
flink-core/src/test/java/org/apache/flink/api/common/operators/base/MapOperatorTest.java
浏览文件 @
65bf092d
...
...
@@ -97,8 +97,8 @@ public class MapOperatorTest implements java.io.Serializable {
parser
,
new
UnaryOperatorInformation
<
String
,
Integer
>(
BasicTypeInfo
.
STRING_TYPE_INFO
,
BasicTypeInfo
.
INT_TYPE_INFO
),
taskName
);
List
<
String
>
input
=
new
ArrayList
<
String
>(
asList
(
"1"
,
"2"
,
"3"
,
"4"
,
"5"
,
"6"
));
List
<
Integer
>
resultMutableSafe
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
),
true
);
List
<
Integer
>
resultRegular
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
),
false
);
List
<
Integer
>
resultMutableSafe
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
,
null
),
true
);
List
<
Integer
>
resultRegular
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
,
null
),
false
);
assertEquals
(
asList
(
1
,
2
,
3
,
4
,
5
,
6
),
resultMutableSafe
);
assertEquals
(
asList
(
1
,
2
,
3
,
4
,
5
,
6
),
resultRegular
);
...
...
flink-core/src/test/java/org/apache/flink/api/common/operators/base/PartitionMapOperatorTest.java
浏览文件 @
65bf092d
...
...
@@ -75,8 +75,8 @@ public class PartitionMapOperatorTest implements java.io.Serializable {
List
<
String
>
input
=
new
ArrayList
<
String
>(
asList
(
"1"
,
"2"
,
"3"
,
"4"
,
"5"
,
"6"
));
List
<
Integer
>
resultMutableSafe
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
),
true
);
List
<
Integer
>
resultRegular
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
),
false
);
List
<
Integer
>
resultMutableSafe
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
,
null
),
true
);
List
<
Integer
>
resultRegular
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
,
null
),
false
);
assertEquals
(
asList
(
1
,
2
,
3
,
4
,
5
,
6
),
resultMutableSafe
);
assertEquals
(
asList
(
1
,
2
,
3
,
4
,
5
,
6
),
resultRegular
);
...
...
flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java
浏览文件 @
65bf092d
...
...
@@ -66,7 +66,7 @@ public class CoGroupOperatorCollectionTest implements Serializable {
.
build
()
);
final
RuntimeContext
ctx
=
new
RuntimeUDFContext
(
"Test UDF"
,
4
,
0
);
final
RuntimeContext
ctx
=
new
RuntimeUDFContext
(
"Test UDF"
,
4
,
0
,
null
);
{
SumCoGroup
udf1
=
new
SumCoGroup
();
...
...
flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
浏览文件 @
65bf092d
...
...
@@ -155,8 +155,8 @@ public class GroupReduceOperatorTest implements java.io.Serializable {
Integer
>(
"foo"
,
3
),
new
Tuple2
<
String
,
Integer
>(
"bar"
,
2
),
new
Tuple2
<
String
,
Integer
>(
"bar"
,
4
)));
List
<
Tuple2
<
String
,
Integer
>>
resultMutableSafe
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
),
true
);
List
<
Tuple2
<
String
,
Integer
>>
resultRegular
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
),
false
);
List
<
Tuple2
<
String
,
Integer
>>
resultMutableSafe
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
,
null
),
true
);
List
<
Tuple2
<
String
,
Integer
>>
resultRegular
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
,
null
),
false
);
Set
<
Tuple2
<
String
,
Integer
>>
resultSetMutableSafe
=
new
HashSet
<
Tuple2
<
String
,
Integer
>>(
resultMutableSafe
);
...
...
flink-java/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
浏览文件 @
65bf092d
...
...
@@ -101,8 +101,8 @@ public class JoinOperatorBaseTest implements Serializable {
));
try
{
List
<
Tuple2
<
Double
,
String
>>
resultSafe
=
base
.
executeOnCollections
(
inputData1
,
inputData2
,
new
RuntimeUDFContext
(
"op"
,
1
,
0
),
true
);
List
<
Tuple2
<
Double
,
String
>>
resultRegular
=
base
.
executeOnCollections
(
inputData1
,
inputData2
,
new
RuntimeUDFContext
(
"op"
,
1
,
0
),
false
);
List
<
Tuple2
<
Double
,
String
>>
resultSafe
=
base
.
executeOnCollections
(
inputData1
,
inputData2
,
new
RuntimeUDFContext
(
"op"
,
1
,
0
,
null
),
true
);
List
<
Tuple2
<
Double
,
String
>>
resultRegular
=
base
.
executeOnCollections
(
inputData1
,
inputData2
,
new
RuntimeUDFContext
(
"op"
,
1
,
0
,
null
),
false
);
assertEquals
(
expected
,
new
HashSet
<
Tuple2
<
Double
,
String
>>(
resultSafe
));
assertEquals
(
expected
,
new
HashSet
<
Tuple2
<
Double
,
String
>>(
resultRegular
));
...
...
flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
浏览文件 @
65bf092d
...
...
@@ -132,8 +132,8 @@ public class ReduceOperatorTest implements java.io.Serializable {
Integer
>(
"foo"
,
3
),
new
Tuple2
<
String
,
Integer
>(
"bar"
,
2
),
new
Tuple2
<
String
,
Integer
>(
"bar"
,
4
)));
List
<
Tuple2
<
String
,
Integer
>>
resultMutableSafe
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
),
true
);
List
<
Tuple2
<
String
,
Integer
>>
resultRegular
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
),
false
);
List
<
Tuple2
<
String
,
Integer
>>
resultMutableSafe
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
,
null
),
true
);
List
<
Tuple2
<
String
,
Integer
>>
resultRegular
=
op
.
executeOnCollections
(
input
,
new
RuntimeUDFContext
(
taskName
,
1
,
0
,
null
),
false
);
Set
<
Tuple2
<
String
,
Integer
>>
resultSetMutableSafe
=
new
HashSet
<
Tuple2
<
String
,
Integer
>>(
resultMutableSafe
);
Set
<
Tuple2
<
String
,
Integer
>>
resultSetRegular
=
new
HashSet
<
Tuple2
<
String
,
Integer
>>(
resultRegular
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录