Page History

...

final OutputTag<String> sideOutput1 = new OutputTag<String>() {}; 

• Add org.apache.flink.util.RichCollector

Update userFunctions using RichCollector(Or MultiCollector) the rationale behind is Collector has been used in lot of places other than stream and transformation, adding direct to Collector interface will introduce many empty methods in those classes.

public interface RichCollector<T> extends Collector<T>{
 <S> void collect(OutputTag<S> tag, S value);
}

FlatMapFunction as example

flatMap(String value, 

...

Collector<Tuple2<String, Integer>> out){

out.collect(new Tuple2<String, Integer>(token, 1));

((RichCollector<Tuple2<String, Integer>>)collector).collect(sideOutput1, "sideout");

}

WindowFunction

public interface WindowFunction<IN, OUT, KEY, W extends Window> extends Function, Serializable{

void apply(KEY key, W window, Iterable<IN> input, 

...

Collector<OUT> out) throws Exception;

}

• Add getCollector method to RuntimeContext

In RichFunction, user can getRuntimeContext().getCollector(), it returns a RichCollector

Add getOutput(OutputTag) to SingleOutputStreamOperator

User may pass outputtag defined earlier here eearlier and get a corresponding outputstream. There can be more than one outputtag share same type.

flatMap(..).getSideOutput(sideOutput1) 

...