> sideoutput might deserve a seperate class which inherit
form singleoutput. It might prevent lot of confusions
Thanks, but how could that be done? Do you mean that if one
calls .process(), then the stream would change to another
class which would only allow calls like .getMainOutput() or
.getSideOutput("name")? Of course compile time error would
be even better than a runtime error, but I don't see yet how
it could be done in practice.
On Sun, Jan 14, 2018 at 4:55 AM, Chen Qin
<qinnc...@gmail.com <mailto:qinnc...@gmail.com>> wrote:
Hi Juho,
I think sideoutput might deserve a seperate class which
inherit form singleoutput. It might prevent lot of
confusions. A more generic question is whether
datastream api can be mulitple ins and mulitple outs
natively. It's more like scheduling problem when you
come from single process system to multiple process
system, which one should get resource and how much
sharing same hardware resources, I guess it will open
gate to lots of edge cases -> strategies-> more edge
cases :)
Chen
On Fri, Jan 12, 2018 at 6:34 AM, Juho Autio
<juho.au...@rovio.com <mailto:juho.au...@rovio.com>> wrote:
Maybe I could express it in a slightly different
way: if adding the .filter() after .process() causes
the side output to be somehow totally "lost", then I
believe the .getSideOutput() could be aware that
there is not such side output to be listened to from
upstream, and throw an exception. I mean, this
should be possible when building the DAG, it
shouldn't require starting the stream to detect?
Thanks..
On Fri, Jan 12, 2018 at 2:48 PM, Tzu-Li (Gordon) Tai
<tzuli...@apache.org <mailto:tzuli...@apache.org>>
wrote:
Hi Juho,
Now that I think of it this seems like a bug to
me: why does the call to getSideOutput succeed
if it doesn't provide _any_ input?
With the way side outputs work, I don’t think
this is possible (or would make sense). An
operator does not know whether or not it would
ever emit some element with a given tag.
As far as I understand it, calling
`getSideOutput` essentially adds a virtual node
to the result stream graph that listens to the
specified tag from the upstream input.
While I’m not aware whether or not your
observation is expected behavior, from an API
perspective, I can see why it is a bit confusing
for you.
Aljoscha would be the expert here, maybe he’ll
have more insights. I’ve looped him in cc’ed.
Cheers,
Gordon
On 12 January 2018 at 4:05:13 PM, Juho Autio
(juho.au...@rovio.com
<mailto:juho.au...@rovio.com>) wrote:
When I run the code below (Flink 1.4.0 or
1.3.1), only "a" is printed. If I switch the
position of .process() & .filter() (ie. filter
first, then process), both "a" & "b" are
printed, as expected.
I guess it's a bit hard to say what the side
output should include in this case: the stream
before filtering or after it?
What I would suggest is Flink to protect
against this kind of a user error that is hard
to debug. Would it be possible that Flink
throws an exception if one tries to call
.getSideOutput() on anything that doesn't
actually provide that side output? Now that I
think of it this seems like a bug to me: why
does the call to getSideOutput succeed if it
doesn't provide _any_ input? I would expect it
to get the side output data stream _before_
applying .filter().
import
org.apache.flink.api.common.functions.FilterFunction;
import
org.apache.flink.streaming.api.datastream.DataStreamSource;
import
org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import
org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
public class SideOutputProblem {
public static void main(String[] args) throws
Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> stream =
env.fromElements("a", "b");
OutputTag<String> sideOutputTag = new
OutputTag<String>("side-output"){};
SingleOutputStreamOperator<String> processed =
stream
.process(new ProcessFunction<String, String>() {
@Override
public void processElement(String s, Context
context, Collector<String> collector) throws
Exception {
if ("a".equals(s)) {
collector.collect(s);
} else {
context.output(sideOutputTag, s);
}
}
})
.filter(new FilterFunction<String>() {
@Override
public boolean filter(String s) throws Exception {
return true;
}
});
processed.getSideOutput(sideOutputTag).printToErr();
processed.print();
env.execute();
}
}
Cheers,
Juho
