[FLINK-4460] Side Outputs in Flink #3484

aljoscha · 2017-03-07T11:10:46Z

This is a refinement of #2982 by @chenqin.

I changed the API a bit, added support for side outputs to ProcessFunction, enabled side outputs to work with chaining, added proper Scala API and a Scala API test and added documentation.

R: @uce @kl0u and @chenqin for review, please

chenqin

Some minor comments, LGTM.
+@uce @kl0u for second opinion

chenqin · 2017-03-08T05:24:32Z

...java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java

+	 * <pre>{@code
+	 * static final OutputTag<X> sideOutputTag = new OutputTag<X>("side-output") {};
+	 *
+	 * public void flatMap(X value, Collector<String> out) throws Exception {


Comments seems out of date, I think we already decided to get ride of CollectorWrapper

chenqin · 2017-03-08T05:31:52Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java

@@ -85,6 +86,7 @@
 	private Set<Integer> sources;
 	private Set<Integer> sinks;
 	private Map<Integer, Tuple2<Integer, List<String>>> virtualSelectNodes;
+	private Map<Integer, Tuple2<Integer, OutputTag>> virtualOutputNodes;


We might consider use addVirtualSideOutputNode and virtualSideOutputNodes. Unless we want to refactor move away from current assumption <IN>operator<OUT> to <<tag1,IN1>...<tagX,INX> operator <<taga,OUTa>...<tagx,OUTX>

The method is already called addVirtualSideOutputNode(). I'm adjusting the name of the field. Thanks!

sounds good

chenqin · 2017-03-08T05:36:52Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java

+			int virtualId = upStreamVertexID;
+			upStreamVertexID = virtualOutputNodes.get(virtualId).f0;
+			if (outputTag == null) {
+				// selections that happen downstream override earlier selections


may consider call out this behavior in getSideOutput comments

I think the comment is a leftover from copying this code from split/select. For side outputs it can't happen that you have multiple "selects" after one another. Will remove the comment. What do you think?

sounds good to me!

chenqin · 2017-03-08T05:38:37Z

...ming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java

@@ -60,6 +60,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import com.google.common.collect.Iterables;


Do you think introduce this dependency is good idea or bad idea? Up to you :)

You're right, I'm changing this to simply have two loops.

I think you introduced this in the first place, though. 😉

That sounds right, good catch!
Thanks for fixing!

chenqin

LGTM

chenqin

LGTM

chenqin

LGTM

chenqin

LGTM

chenqin

If no objection from @uce @kl0u, please land this diff

chenqin · 2017-03-08T19:12:29Z

...ming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java

@@ -60,6 +60,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import com.google.common.collect.Iterables;


That sounds right, good catch!
Thanks for fixing!

uce · 2017-03-09T10:24:11Z

Unfortunately, I won't have time to look over this PR this week. Thanks for pinging me @aljoscha @chenqin.

kl0u · 2017-03-09T17:42:50Z

...ava/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java

+		// element not handled by any window
+		// late arriving tag has been set
+		// windowAssigner is event time and current timestamp + allowed lateness no less than element timestamp
+		if(isSkippedElement && lateDataOutputTag != null && isLate(element)) {


@chenqin and @aljoscha I am starting to review the PR and I was wondering when is this new isLate() check needed? At least for the out-of-box window assigners, this seems to be a redundant check.

Thanks @kl0u Good catch!

I put isLate there with intention to filter out dropped events with other reasons which I may not aware of. lateArrivingEvents is really late arriving and dropped events.

@aljoscha If that is redundant check, we might just remove isLate.
What do you think?

I thought about this again. I think it doesn't hurt to have it because it catches the case when a WindowAssigner doesn't assign any windows. In that case an element is also "skipped" but it is not necessarily considered late. What do you think?

I just added a test for the behaviour with a "weird" WindowAssigner.

kl0u

I had some comments, most on code style and duplicate code. There is one that raises some correctness concerns. It is in the OutputTag class, in the equals method, the second point.

This review was just the initial one. I may have some additional comments.

kl0u · 2017-03-10T13:37:25Z

flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeHint.java

+		this.typeInfo = TypeExtractor.createTypeInfo(instance, baseClass, instance.getClass(), genericParameterPos);
+	}
+
+


Remove one of the 2 empty lines.

kl0u · 2017-03-10T13:38:44Z

flink-core/src/main/java/org/apache/flink/util/OutputTag.java

+	 * @param id The id of the created {@code OutputTag}.
+     */
+	public OutputTag(String id) {
+		Preconditions.checkNotNull(id, "OutputTag id cannot be null.");


We do not need both lines with the checks. We can just have:

this.id = Preconditions.checkNotNull(id, "OutputTag id cannot be null.");

kl0u · 2017-03-10T13:39:31Z

flink-core/src/main/java/org/apache/flink/util/OutputTag.java

+		Preconditions.checkNotNull(id, "OutputTag id cannot be null.");
+		this.id = requireNonNull(id);
+
+		try {


No need for line breaking:
TypeHint<T> typeHint = new TypeHint<T>(OutputTag.class, this, 0) {};

kl0u · 2017-03-10T13:39:58Z

flink-core/src/main/java/org/apache/flink/util/OutputTag.java

+	 */
+	public OutputTag(String id, TypeInformation<T> typeInfo) {
+		this.id = Preconditions.checkNotNull(id, "OutputTag id cannot be null.");
+		this.typeInfo =


No need for line breaking.

kl0u · 2017-03-10T13:42:30Z

flink-core/src/main/java/org/apache/flink/util/OutputTag.java

+
+	@Override
+	public boolean equals(Object obj) {
+		return obj instanceof OutputTag


Two points:

we cannot have this.id == null or (OutputTag) obj).id == null because we check at the constructor, so this method can be simplified.

we never check for uniqueness of the outputTag.id. We should do it at the translation. This is also a correctness issue as this may result in undesired sideoutput "collisions.

I would have liked to include the TypeInformation into the check but we can't do that because it's transient. I'll try and figure something out for checking that side outputs are unique, not as easy as it seems.

I see. The problem is that if this does not work, then we can have important side effects.

Still the first comment applies: the equals can be simplified given that id != null.

kl0u · 2017-03-10T14:09:07Z

...src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorTest.java

@@ -1528,14 +1572,16 @@ public void testDropDueToLatenessSessionZeroLatenessPurgingTrigger() throws Exce
 				stateDesc,
 				new InternalSingleValueWindowFunction<>(new ReducedSessionWindowFunction()),
 				PurgingTrigger.of(EventTimeTrigger.create()),
-				LATENESS);
+				LATENESS,
+					lateOutputTag);


wrong alignment

kl0u · 2017-03-10T14:09:29Z

...src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorTest.java

@@ -1618,14 +1664,16 @@ public void testDropDueToLatenessSessionZeroLateness() throws Exception {
 				stateDesc,
 				new InternalSingleValueWindowFunction<>(new ReducedSessionWindowFunction()),
 				EventTimeTrigger.create(),
-				LATENESS);
+				LATENESS,


wrong alignment

kl0u · 2017-03-10T14:09:43Z

...src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperatorTest.java

@@ -1702,15 +1754,16 @@ public void testDropDueToLatenessSessionWithLatenessPurgingTrigger() throws Exce
 				stateDesc,
 				new InternalSingleValueWindowFunction<>(new ReducedSessionWindowFunction()),
 				PurgingTrigger.of(EventTimeTrigger.create()),
-				LATENESS);
+				LATENESS,


wrong alignment

kl0u · 2017-03-10T14:11:32Z

flink-streaming-java/src/test/java/org/apache/flink/streaming/util/CollectorOutput.java

@@ -53,5 +54,11 @@ public void collect(StreamRecord<T> record) {
 	}

 	@Override


The signature can go on the same line

kl0u · 2017-03-10T14:12:16Z

flink-streaming-java/src/test/java/org/apache/flink/streaming/util/MockOutput.java

@@ -40,6 +41,12 @@ public void collect(StreamRecord<T> record) {
 	}

 	@Override
+	public <X> void collect(


The signature fits in one line

aljoscha · 2017-03-10T16:53:58Z

Thanks @kl0u for the (already) quite thorough review! I'll push a commit with fixes.

aljoscha · 2017-03-11T07:54:10Z

@kl0u @chenqin I cleaned up the commits, distributed the fixes from the comments to the right commits. I also added more tests/ITCases for: detecting name clashes in side output IDs, side outputs with multiple consumers.

kl0u · 2017-03-11T08:01:56Z

Thanks @aljoscha I will have a look on Monday.

kl0u

Hi @aljoscha . I did a first pass which did not include the translation part of the StreamGraph. I will continue with this part now.

So far I had some minor comments and one more important at the CopyingChainingOutput.pushToOperator().

kl0u · 2017-03-15T11:04:03Z

flink-core/src/main/java/org/apache/flink/util/OutputTag.java

+
+	@Override
+	public boolean equals(Object obj) {
+		return obj instanceof OutputTag


Still the first comment applies: the equals can be simplified given that id != null.

kl0u · 2017-03-15T11:04:55Z

...java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java

+	 * into the side output with the given {@link OutputTag}.
+	 *
+	 * @see org.apache.flink.streaming.api.functions.ProcessFunction.Context#output(OutputTag, Object) 
+	 */


Missing space between the ) and the {

kl0u · 2017-03-15T11:05:22Z

...java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java

+	 * @see org.apache.flink.streaming.api.functions.ProcessFunction.Context#output(OutputTag, Object) 
+	 */
+	public <X> DataStream<X> getSideOutput(OutputTag<X> sideOutputTag){
+		sideOutputTag = clean(sideOutputTag);


I think it is better to not reuse the argument variable but create a new one.

kl0u · 2017-03-15T11:06:02Z

...java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java

+		}
+
+		requestedSideOutputs.put(sideOutputTag, sideOutputTag.getTypeInfo());
+


The requireNotNull should be in the beginning of the method.

kl0u · 2017-03-15T11:08:38Z

...a/src/main/java/org/apache/flink/streaming/api/transformations/SideOutputTransformation.java

+ * connected to downstream operations.
+ *
+ * @param <T> The type of the elements that result from this {@code SideOutputTransformation}
+ */


Here we do not check if the input is null (we do it in the caller method only) but we try get the parallelism. We could have the parallelism as a separate argument, and then, after the super() check if the input is null.
This makes the code of the class self-contained as you do not have to check other classes to see if the input can be null or not. What do you think?

kl0u · 2017-03-15T11:09:36Z

...ava/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java

+	 * @param element The element to check
+	 * @return The element for which should be considered when sideoutputs
+	 */
+	protected boolean isLate(StreamRecord<IN> element){


This is not used any more, right? So it can be deleted.

I must have removed the check by accident. I think we agreed to rename this to something more meaningful and keep it, right?

That is what I remember as well.

kl0u · 2017-03-15T11:10:51Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java

+			outSerializer = upStreamConfig.getTypeSerializerSideOut(
+					edge.getOutputTag(), taskEnvironment.getUserClassLoader());
+		} else {
+			// main output


this can become one line.

kl0u · 2017-03-15T11:11:40Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java

+
+			pushToOperator(record);
+		}
+


This can become private as the copying alternative has its own implementation, right?

kl0u · 2017-03-15T11:12:13Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java

+			pushToOperator(record);
+		}
+
+		@Override


This can become private, as before.

kl0u · 2017-03-15T11:12:39Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OperatorChain.java

-				operator.processElement(copy);
-			}
-			catch (Exception e) {
+				operator.processElement(castRecord);


This should be copy, not castRecord.

kl0u

@aljoscha I finished my review. I had some comments.

kl0u · 2017-03-15T13:23:04Z

...treaming/src/main/java/org/apache/flink/streaming/examples/sideoutput/SideOutputExample.java

+	 * We need to create an {@link OutputTag} so that we can reference it when emitting
+	 * data to a side output and also to retrieve the side output stream from an operation.
+	 */
+	static final OutputTag<String> rejectedWordsTag = new OutputTag<String>("rejected") {};


Here we add a side output but we do nothing to show that it works. Probably we can add a prefix "rejected-" to the record and print it, so that the user can see what the side output does.

kl0u · 2017-03-15T13:24:10Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamEdge.java

 		this.sourceVertex = sourceVertex;
 		this.targetVertex = targetVertex;
 		this.typeNumber = typeNumber;
 		this.selectedNames = selectedNames;
 		this.outputPartitioner = outputPartitioner;
+		this.outputTag = outputTag;



Does it make sense to add the outputTag also in the edgeId?

Not sure what the edge id exactly does and who uses it so I prefer to not touch it, for now.

kl0u · 2017-03-15T13:25:39Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java

@@ -333,32 +373,39 @@ public void addEdge(Integer upStreamVertexID, Integer downStreamVertexID, int ty
 				downStreamVertexID,
 				typeNumber,
 				null,
-				new ArrayList<String>());
+				new ArrayList<String>(), null);


The null should go to the next line for uniformity.

kl0u · 2017-03-15T13:27:51Z

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java

@@ -63,6 +66,7 @@
 	private TypeSerializer<?> typeSerializerIn1;
 	private TypeSerializer<?> typeSerializerIn2;
 	private TypeSerializer<?> typeSerializerOut;
+	private Map<OutputTag<?>, TypeSerializer<?>> typeSerializerMap;


This is not used anywhere in the code. Can it be removed, along with the getTypeSerializerOut() and setTypeSerializerOut()?

aljoscha · 2017-03-17T16:21:10Z

Thanks for reviewing again, @kl0u! I incorporated all your suggestions. I'm now waiting for travis to give the green light and then I'll merge.

@chenqin A lot of thanks also to you for working on this and pushing it with me! 😃

This does not yet allow users to emit to side outputs in user functions. Only operators (StreamOperator) can emit to side outputs. A side output can be retrieved on a SingleOutputStreamOperator.

We use side outputs to emit dropped late data.

The Scala side output API uses context bounds to get a TypeInformation for an OutputTag. This also adds a SideOutputITCase for the Scala API.

This also adds tests.

aljoscha force-pushed the finish-pr-2982-side-outputs-cp branch 2 times, most recently from 1b70ee5 to 8c5ac2e Compare March 7, 2017 15:37

chenqin reviewed Mar 8, 2017

View reviewed changes

chenqin approved these changes Mar 8, 2017

View reviewed changes

chenqin reviewed Mar 8, 2017

View reviewed changes

chenqin approved these changes Mar 8, 2017

View reviewed changes

chenqin approved these changes Mar 9, 2017

View reviewed changes

kl0u reviewed Mar 9, 2017

View reviewed changes

kl0u requested changes Mar 10, 2017

View reviewed changes

aljoscha force-pushed the finish-pr-2982-side-outputs-cp branch from 3c521b2 to efefb83 Compare March 11, 2017 07:53

kl0u reviewed Mar 15, 2017

View reviewed changes

kl0u requested changes Mar 15, 2017

View reviewed changes

aljoscha force-pushed the finish-pr-2982-side-outputs-cp branch from efefb83 to 62cc5ee Compare March 17, 2017 16:19

aljoscha force-pushed the finish-pr-2982-side-outputs-cp branch from 62cc5ee to 20d8d67 Compare March 17, 2017 20:32

chenqin and others added 3 commits March 17, 2017 21:36

[FLINK-4460] Add support for side outputs

d6105f6

This does not yet allow users to emit to side outputs in user functions. Only operators (StreamOperator) can emit to side outputs. A side output can be retrieved on a SingleOutputStreamOperator.

[FLINK-4460] Provide late-data output for window operations

9d3d16f

We use side outputs to emit dropped late data.

[FLINK-4460] Expose OutputTag constructor that takes TypeInformation

dd32a3c

aljoscha added 8 commits March 17, 2017 21:36

[FLINK-4460] Make chaining work with side outputs

a5d69f5

[FLINK-4460] Add side outputs for ProcessFunction

198a933

[FLINK-4460] Add proper side output API for Scala API

ec46231

The Scala side output API uses context bounds to get a TypeInformation for an OutputTag. This also adds a SideOutputITCase for the Scala API.

[FLINK-4460] Add documentation for side outputs

64d639a

[FLINK-4460] Add WindowOperatorContractTest tests for late data emission

a2c79a1

[FLINK-4460] Add ITCase to verify watermark forwarding for side outputs

3bae526

[FLINK-4460] Add side output ITCase for multiple side out consumers

1eb7bab

[FLINK-4460] Catch side output OutputTag id clashes

d0eef93

This also adds tests.

aljoscha force-pushed the finish-pr-2982-side-outputs-cp branch from 20d8d67 to d0eef93 Compare March 17, 2017 20:36

aljoscha closed this Mar 18, 2017

aljoscha deleted the finish-pr-2982-side-outputs-cp branch March 18, 2017 07:14

rmetzger added the component=API/DataStream label Mar 14, 2019

		this.typeInfo = TypeExtractor.createTypeInfo(instance, baseClass, instance.getClass(), genericParameterPos);
		}

		@@ -53,5 +54,11 @@ public void collect(StreamRecord<T> record) {
		}

		@Override

		}

		requestedSideOutputs.put(sideOutputTag, sideOutputTag.getTypeInfo());

[FLINK-4460] Side Outputs in Flink #3484

[FLINK-4460] Side Outputs in Flink #3484

Conversation

aljoscha commented Mar 7, 2017

chenqin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chenqin left a comment

Choose a reason for hiding this comment

chenqin left a comment

Choose a reason for hiding this comment

chenqin left a comment

Choose a reason for hiding this comment

chenqin left a comment

Choose a reason for hiding this comment

chenqin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

uce commented Mar 9, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kl0u left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aljoscha commented Mar 10, 2017

aljoscha commented Mar 11, 2017

kl0u commented Mar 11, 2017

kl0u left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kl0u left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aljoscha commented Mar 17, 2017