Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
doujutun3207
flink
提交
a0109e17
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,发现更多精彩内容 >>
提交
a0109e17
编写于
3月 11, 2017
作者:
A
Aljoscha Krettek
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[FLINK-4460] Catch side output OutputTag id clashes
This also adds tests.
上级
709edb80
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
105 addition
and
54 deletion
+105
-54
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java
.../streaming/api/datastream/SingleOutputStreamOperator.java
+26
-3
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
...ava/org/apache/flink/streaming/api/graph/StreamGraph.java
+19
-1
flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SideOutputITCase.java
...apache/flink/test/streaming/runtime/SideOutputITCase.java
+60
-50
未找到文件。
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java
浏览文件 @
a0109e17
...
...
@@ -33,6 +33,9 @@ import org.apache.flink.streaming.api.transformations.StreamTransformation;
import
org.apache.flink.streaming.runtime.partitioner.StreamPartitioner
;
import
org.apache.flink.util.Preconditions
;
import
java.util.HashMap
;
import
java.util.Map
;
import
static
java
.
util
.
Objects
.
requireNonNull
;
/**
...
...
@@ -47,6 +50,13 @@ public class SingleOutputStreamOperator<T> extends DataStream<T> {
/** Indicate this is a non-parallel operator and cannot set a non-1 degree of parallelism. **/
protected
boolean
nonParallel
=
false
;
/**
* We keep track of the side outputs that were already requested and their types. With this,
* we can catch the case when a side output with a matching id is requested for a different
* type because this would lead to problems at runtime.
*/
private
Map
<
OutputTag
<?>,
TypeInformation
>
requestedSideOutputs
=
new
HashMap
<>();
protected
SingleOutputStreamOperator
(
StreamExecutionEnvironment
environment
,
StreamTransformation
<
T
>
transformation
)
{
super
(
environment
,
transformation
);
}
...
...
@@ -425,9 +435,22 @@ public class SingleOutputStreamOperator<T> extends DataStream<T> {
*
* @see org.apache.flink.streaming.api.functions.ProcessFunction.Context#output(OutputTag, Object)
*/
public
<
X
>
DataStream
<
X
>
getSideOutput
(
OutputTag
<
X
>
sideOutputTag
){
sideOutputTag
=
clean
(
sideOutputTag
);
SideOutputTransformation
<
X
>
sideOutputTransformation
=
new
SideOutputTransformation
<>(
this
.
getTransformation
(),
requireNonNull
(
sideOutputTag
));
public
<
X
>
DataStream
<
X
>
getSideOutput
(
OutputTag
<
X
>
sideOutputTag
)
{
sideOutputTag
=
clean
(
requireNonNull
(
sideOutputTag
));
// make a defensive copy
sideOutputTag
=
new
OutputTag
<
X
>(
sideOutputTag
.
getId
(),
sideOutputTag
.
getTypeInfo
());
TypeInformation
<?>
type
=
requestedSideOutputs
.
get
(
sideOutputTag
);
if
(
type
!=
null
&&
!
type
.
equals
(
sideOutputTag
.
getTypeInfo
()))
{
throw
new
UnsupportedOperationException
(
"A side output with a matching id was "
+
"already requested with a different type. This is not allowed, side output "
+
"ids need to be unique."
);
}
requestedSideOutputs
.
put
(
sideOutputTag
,
sideOutputTag
.
getTypeInfo
());
SideOutputTransformation
<
X
>
sideOutputTransformation
=
new
SideOutputTransformation
<>(
this
.
getTransformation
(),
sideOutputTag
);
return
new
DataStream
<>(
this
.
getExecutionEnvironment
(),
sideOutputTransformation
);
}
}
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
浏览文件 @
a0109e17
...
...
@@ -310,6 +310,23 @@ public class StreamGraph extends StreamingPlan {
throw
new
IllegalStateException
(
"Already has virtual output node with id "
+
virtualId
);
}
// verify that we don't already have a virtual node for the given originalId/outputTag
// combination with a different TypeInformation. This would indicate that someone is trying
// to read a side output from an operation with a different type for the same side output
// id.
for
(
Tuple2
<
Integer
,
OutputTag
>
tag
:
virtualSideOutputNodes
.
values
())
{
if
(!
tag
.
f0
.
equals
(
originalId
))
{
// different source operator
continue
;
}
if
(!
tag
.
f1
.
getTypeInfo
().
equals
(
outputTag
.
getTypeInfo
()))
{
throw
new
IllegalArgumentException
(
"Trying to add a side input for the same id "
+
"with a different type. This is not allowed."
);
}
}
virtualSideOutputNodes
.
put
(
virtualId
,
new
Tuple2
<>(
originalId
,
outputTag
));
}
...
...
@@ -358,7 +375,8 @@ public class StreamGraph extends StreamingPlan {
downStreamVertexID
,
typeNumber
,
null
,
new
ArrayList
<
String
>(),
null
);
new
ArrayList
<
String
>(),
null
);
}
...
...
flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SideOutputITCase.java
浏览文件 @
a0109e17
...
...
@@ -40,7 +40,9 @@ import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import
org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase
;
import
org.apache.flink.test.streaming.runtime.util.TestListResultSink
;
import
org.apache.flink.util.Collector
;
import
org.junit.Rule
;
import
org.junit.Test
;
import
org.junit.rules.ExpectedException
;
import
javax.annotation.Nullable
;
import
java.io.Serializable
;
...
...
@@ -56,6 +58,9 @@ import static org.junit.Assert.assertEquals;
*/
public
class
SideOutputITCase
extends
StreamingMultipleProgramsTestBase
implements
Serializable
{
@Rule
public
transient
ExpectedException
expectedException
=
ExpectedException
.
none
();
static
List
<
Integer
>
elements
=
new
ArrayList
<>();
static
{
elements
.
add
(
1
);
...
...
@@ -65,14 +70,14 @@ public class SideOutputITCase extends StreamingMultipleProgramsTestBase implemen
elements
.
add
(
4
);
}
private
final
static
OutputTag
<
String
>
sideOutputTag1
=
new
OutputTag
<
String
>(
"side"
){};
private
final
static
OutputTag
<
String
>
sideOutputTag2
=
new
OutputTag
<
String
>(
"other-side"
){};
/**
* Verify that watermarks are forwarded to all side outputs.
*/
@Test
public
void
testWatermarkForwarding
()
throws
Exception
{
final
OutputTag
<
String
>
sideOutputTag1
=
new
OutputTag
<
String
>(
"side"
){};
final
OutputTag
<
String
>
sideOutputTag2
=
new
OutputTag
<
String
>(
"other-side"
){};
TestListResultSink
<
String
>
sideOutputResultSink1
=
new
TestListResultSink
<>();
TestListResultSink
<
String
>
sideOutputResultSink2
=
new
TestListResultSink
<>();
TestListResultSink
<
String
>
resultSink
=
new
TestListResultSink
<>();
...
...
@@ -167,6 +172,8 @@ public class SideOutputITCase extends StreamingMultipleProgramsTestBase implemen
@Test
public
void
testSideOutputWithMultipleConsumers
()
throws
Exception
{
final
OutputTag
<
String
>
sideOutputTag
=
new
OutputTag
<
String
>(
"side"
){};
TestListResultSink
<
String
>
sideOutputResultSink1
=
new
TestListResultSink
<>();
TestListResultSink
<
String
>
sideOutputResultSink2
=
new
TestListResultSink
<>();
TestListResultSink
<
Integer
>
resultSink
=
new
TestListResultSink
<>();
...
...
@@ -184,12 +191,12 @@ public class SideOutputITCase extends StreamingMultipleProgramsTestBase implemen
public
void
processElement
(
Integer
value
,
Context
ctx
,
Collector
<
Integer
>
out
)
throws
Exception
{
out
.
collect
(
value
);
ctx
.
output
(
sideOutputTag
1
,
"sideout-"
+
String
.
valueOf
(
value
));
ctx
.
output
(
sideOutputTag
,
"sideout-"
+
String
.
valueOf
(
value
));
}
});
passThroughtStream
.
getSideOutput
(
sideOutputTag
1
).
addSink
(
sideOutputResultSink1
);
passThroughtStream
.
getSideOutput
(
sideOutputTag
1
).
addSink
(
sideOutputResultSink2
);
passThroughtStream
.
getSideOutput
(
sideOutputTag
).
addSink
(
sideOutputResultSink1
);
passThroughtStream
.
getSideOutput
(
sideOutputTag
).
addSink
(
sideOutputResultSink2
);
passThroughtStream
.
addSink
(
resultSink
);
env
.
execute
();
...
...
@@ -200,6 +207,8 @@ public class SideOutputITCase extends StreamingMultipleProgramsTestBase implemen
@Test
public
void
testSideOutputWithMultipleConsumersWithObjectReuse
()
throws
Exception
{
final
OutputTag
<
String
>
sideOutputTag
=
new
OutputTag
<
String
>(
"side"
){};
TestListResultSink
<
String
>
sideOutputResultSink1
=
new
TestListResultSink
<>();
TestListResultSink
<
String
>
sideOutputResultSink2
=
new
TestListResultSink
<>();
TestListResultSink
<
Integer
>
resultSink
=
new
TestListResultSink
<>();
...
...
@@ -218,12 +227,12 @@ public class SideOutputITCase extends StreamingMultipleProgramsTestBase implemen
public
void
processElement
(
Integer
value
,
Context
ctx
,
Collector
<
Integer
>
out
)
throws
Exception
{
out
.
collect
(
value
);
ctx
.
output
(
sideOutputTag
1
,
"sideout-"
+
String
.
valueOf
(
value
));
ctx
.
output
(
sideOutputTag
,
"sideout-"
+
String
.
valueOf
(
value
));
}
});
passThroughtStream
.
getSideOutput
(
sideOutputTag
1
).
addSink
(
sideOutputResultSink1
);
passThroughtStream
.
getSideOutput
(
sideOutputTag
1
).
addSink
(
sideOutputResultSink2
);
passThroughtStream
.
getSideOutput
(
sideOutputTag
).
addSink
(
sideOutputResultSink1
);
passThroughtStream
.
getSideOutput
(
sideOutputTag
).
addSink
(
sideOutputResultSink2
);
passThroughtStream
.
addSink
(
resultSink
);
env
.
execute
();
...
...
@@ -232,13 +241,13 @@ public class SideOutputITCase extends StreamingMultipleProgramsTestBase implemen
assertEquals
(
Arrays
.
asList
(
1
,
2
,
3
,
4
,
5
),
resultSink
.
getSortedResult
());
}
/**
* Test ProcessFunction side output.
*/
@Test
public
void
testProcessFunctionSideOutput
()
throws
Exception
{
TestListResultSink
<
String
>
sideOutputResultSink
=
new
TestListResultSink
<>();
TestListResultSink
<
Integer
>
resultSink
=
new
TestListResultSink
<>();
public
void
testSideOutputNameClash
()
throws
Exception
{
final
OutputTag
<
String
>
sideOutputTag1
=
new
OutputTag
<
String
>(
"side"
){};
final
OutputTag
<
Integer
>
sideOutputTag2
=
new
OutputTag
<
Integer
>(
"side"
){};
TestListResultSink
<
String
>
sideOutputResultSink1
=
new
TestListResultSink
<>();
TestListResultSink
<
Integer
>
sideOutputResultSink2
=
new
TestListResultSink
<>();
StreamExecutionEnvironment
see
=
StreamExecutionEnvironment
.
getExecutionEnvironment
();
see
.
setParallelism
(
3
);
...
...
@@ -254,22 +263,23 @@ public class SideOutputITCase extends StreamingMultipleProgramsTestBase implemen
Integer
value
,
Context
ctx
,
Collector
<
Integer
>
out
)
throws
Exception
{
out
.
collect
(
value
);
ctx
.
output
(
sideOutputTag1
,
"sideout-"
+
String
.
valueOf
(
value
));
ctx
.
output
(
sideOutputTag2
,
13
);
}
});
passThroughtStream
.
getSideOutput
(
sideOutputTag1
).
addSink
(
sideOutputResultSink
);
passThroughtStream
.
addSink
(
resultSink
);
see
.
execute
();
passThroughtStream
.
getSideOutput
(
sideOutputTag1
).
addSink
(
sideOutputResultSink1
);
assertEquals
(
Arrays
.
asList
(
"sideout-1"
,
"sideout-2"
,
"sideout-3"
,
"sideout-4"
,
"sideout-5"
),
sideOutputResultSink
.
getSortedResult
()
);
assertEquals
(
Arrays
.
asList
(
1
,
2
,
3
,
4
,
5
),
resultSink
.
getSortedResult
()
);
expectedException
.
expect
(
UnsupportedOperationException
.
class
);
passThroughtStream
.
getSideOutput
(
sideOutputTag2
).
addSink
(
sideOutputResultSink2
);
}
/**
* Test
keyed
ProcessFunction side output.
* Test ProcessFunction side output.
*/
@Test
public
void
testKeyedProcessFunctionSideOutput
()
throws
Exception
{
public
void
testProcessFunctionSideOutput
()
throws
Exception
{
final
OutputTag
<
String
>
sideOutputTag
=
new
OutputTag
<
String
>(
"side"
){};
TestListResultSink
<
String
>
sideOutputResultSink
=
new
TestListResultSink
<>();
TestListResultSink
<
Integer
>
resultSink
=
new
TestListResultSink
<>();
...
...
@@ -279,14 +289,6 @@ public class SideOutputITCase extends StreamingMultipleProgramsTestBase implemen
DataStream
<
Integer
>
dataStream
=
see
.
fromCollection
(
elements
);
SingleOutputStreamOperator
<
Integer
>
passThroughtStream
=
dataStream
.
keyBy
(
new
KeySelector
<
Integer
,
Integer
>()
{
private
static
final
long
serialVersionUID
=
1L
;
@Override
public
Integer
getKey
(
Integer
value
)
throws
Exception
{
return
value
;
}
})
.
process
(
new
ProcessFunction
<
Integer
,
Integer
>()
{
private
static
final
long
serialVersionUID
=
1L
;
...
...
@@ -294,11 +296,11 @@ public class SideOutputITCase extends StreamingMultipleProgramsTestBase implemen
public
void
processElement
(
Integer
value
,
Context
ctx
,
Collector
<
Integer
>
out
)
throws
Exception
{
out
.
collect
(
value
);
ctx
.
output
(
sideOutputTag
1
,
"sideout-"
+
String
.
valueOf
(
value
));
ctx
.
output
(
sideOutputTag
,
"sideout-"
+
String
.
valueOf
(
value
));
}
});
passThroughtStream
.
getSideOutput
(
sideOutputTag
1
).
addSink
(
sideOutputResultSink
);
passThroughtStream
.
getSideOutput
(
sideOutputTag
).
addSink
(
sideOutputResultSink
);
passThroughtStream
.
addSink
(
resultSink
);
see
.
execute
();
...
...
@@ -306,20 +308,30 @@ public class SideOutputITCase extends StreamingMultipleProgramsTestBase implemen
assertEquals
(
Arrays
.
asList
(
1
,
2
,
3
,
4
,
5
),
resultSink
.
getSortedResult
());
}
/**
* Test
ProcessFunction side outputs with wrong {@code OutputTag}
.
* Test
keyed ProcessFunction side output
.
*/
@Test
public
void
testProcessFunctionSideOutputWithWrongTag
()
throws
Exception
{
public
void
testKeyedProcessFunctionSideOutput
()
throws
Exception
{
final
OutputTag
<
String
>
sideOutputTag
=
new
OutputTag
<
String
>(
"side"
){};
TestListResultSink
<
String
>
sideOutputResultSink
=
new
TestListResultSink
<>();
TestListResultSink
<
Integer
>
resultSink
=
new
TestListResultSink
<>();
StreamExecutionEnvironment
see
=
StreamExecutionEnvironment
.
getExecutionEnvironment
();
see
.
setParallelism
(
3
);
DataStream
<
Integer
>
dataStream
=
see
.
fromCollection
(
elements
);
dataStream
SingleOutputStreamOperator
<
Integer
>
passThroughtStream
=
dataStream
.
keyBy
(
new
KeySelector
<
Integer
,
Integer
>()
{
private
static
final
long
serialVersionUID
=
1L
;
@Override
public
Integer
getKey
(
Integer
value
)
throws
Exception
{
return
value
;
}
})
.
process
(
new
ProcessFunction
<
Integer
,
Integer
>()
{
private
static
final
long
serialVersionUID
=
1L
;
...
...
@@ -327,20 +339,27 @@ public class SideOutputITCase extends StreamingMultipleProgramsTestBase implemen
public
void
processElement
(
Integer
value
,
Context
ctx
,
Collector
<
Integer
>
out
)
throws
Exception
{
out
.
collect
(
value
);
ctx
.
output
(
sideOutputTag
2
,
"sideout-"
+
String
.
valueOf
(
value
));
ctx
.
output
(
sideOutputTag
,
"sideout-"
+
String
.
valueOf
(
value
));
}
})
.
getSideOutput
(
sideOutputTag1
).
addSink
(
sideOutputResultSink
)
;
});
passThroughtStream
.
getSideOutput
(
sideOutputTag
).
addSink
(
sideOutputResultSink
);
passThroughtStream
.
addSink
(
resultSink
);
see
.
execute
();
assertEquals
(
Arrays
.
asList
(),
sideOutputResultSink
.
getSortedResult
());
assertEquals
(
Arrays
.
asList
(
"sideout-1"
,
"sideout-2"
,
"sideout-3"
,
"sideout-4"
,
"sideout-5"
),
sideOutputResultSink
.
getSortedResult
());
assertEquals
(
Arrays
.
asList
(
1
,
2
,
3
,
4
,
5
),
resultSink
.
getSortedResult
());
}
/**
* Test
keyed
ProcessFunction side outputs with wrong {@code OutputTag}.
* Test ProcessFunction side outputs with wrong {@code OutputTag}.
*/
@Test
public
void
testKeyedProcessFunctionSideOutputWithWrongTag
()
throws
Exception
{
public
void
testProcessFunctionSideOutputWithWrongTag
()
throws
Exception
{
final
OutputTag
<
String
>
sideOutputTag1
=
new
OutputTag
<
String
>(
"side"
){};
final
OutputTag
<
String
>
sideOutputTag2
=
new
OutputTag
<
String
>(
"other-side"
){};
TestListResultSink
<
String
>
sideOutputResultSink
=
new
TestListResultSink
<>();
StreamExecutionEnvironment
see
=
StreamExecutionEnvironment
.
getExecutionEnvironment
();
...
...
@@ -349,14 +368,6 @@ public class SideOutputITCase extends StreamingMultipleProgramsTestBase implemen
DataStream
<
Integer
>
dataStream
=
see
.
fromCollection
(
elements
);
dataStream
.
keyBy
(
new
KeySelector
<
Integer
,
Integer
>()
{
private
static
final
long
serialVersionUID
=
1L
;
@Override
public
Integer
getKey
(
Integer
value
)
throws
Exception
{
return
value
;
}
})
.
process
(
new
ProcessFunction
<
Integer
,
Integer
>()
{
private
static
final
long
serialVersionUID
=
1L
;
...
...
@@ -373,7 +384,6 @@ public class SideOutputITCase extends StreamingMultipleProgramsTestBase implemen
assertEquals
(
Arrays
.
asList
(),
sideOutputResultSink
.
getSortedResult
());
}
private
static
class
TestWatermarkAssigner
implements
AssignerWithPunctuatedWatermarks
<
Integer
>
{
private
static
final
long
serialVersionUID
=
1L
;
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录