https://flink.apache.org/ logo
Join Slack
Powered by
Question about side outputs and OutputTags in pyfl...
# troubleshooting
a
Question about side outputs and OutputTags in pyflink. The docs say we are supposed to 
yield output_tag, value
. But, when I do that, I’m just getting a Tuple(output_tag, value) in the main datastream, not the side output one. What am I doing wrong?
d
Could you show some code snippet? It could help digging into the problem.
a
Hm, let me try to reproduce outside of my code with a basic example…
Okay, I figured it out, but it is very unexpected.
If I do not immediately call 
ds.get_side_output(output_tag)
after 
ds.process(…)
, the elements to the side output in the ProcessFunction are not removed from the main DataStream.
my problem was that I was calling get_side_output too late, and the Tuple(output_tag, value) records were being processed by the main datastream pipeline.
This is very strange. I’d expect that the act of yielding a Tuple(output_tag, value) from the ProcessFunction would mean that that record is no longer in the main datastream.
Here’s an example using word_count.py
Emailed the user mailing list to see if we just need a doc change
d
@Andrew Otto Hey, 
ds.get_side_output(output_tag)
doesn’t have to happen after 
ds.process(…)
immediately. For example, you could move it the end of the file as following:
Copy code
side_output_ds = processed_ds.get_side_output(output_tag)
    if side_output_ds:
        print("printing side output to stdout")
        side_output_ds.print()
However, you must define ``ds.get_side_output(output_tag)`` if it emits side output in the process function.
a
yes ^ I guess that is the point. By the way the doc reads, I would have expected that to be optional. I would have expected that yielding the ouput_tag, value would be enough to keep it out of the main data stream
Just saw your response on email list. Thank you!
Can I emit to the same output_tag in different ProcessFunction process calls, and get all the records emitted into that output tag into the same datastream?
e.g.
datastream.process(ProcessFunc1()).process(ProcessFunc2())
where both ProcessFunc1 and ProcessFunc2 sometimes yield to the same output_tag?
I’m trying to implement a generic error event handler, and I want to catch errors in process functions and emit them all to the same sideoutput datastream
d
@Andrew Otto Hey, you could use the same output_tag, however, you have to get each side output and union them to add some generic error processing.
a
Okay, cool. I don’t need any error processing in my app, I just need to put them all in the same sink. I might be able to do datastream.get_side_ouput().to_sink(…) each time. I’ll try that.
thtank you!
Hm, i’m not so sure I can use the same output_tag in two different ProcessFunctions
doing so causes a Pickle error, i think because using the same output_tag ref in two different functions means that there has to be some locking
TypeError: cannot pickle '_thread.RLock' object
I sent this question to the mailling list.
d
@Andrew Otto Juntao Hu has replied this question in the mailing list. PS: Suggest to post one problem to only one place to avoid having to reply to the same issue twice.
a
Meaning here vs mailing list? Yeah I probably should have just done mailing list. Sorry, thanks.
Thank you for your help @Dian Fu!
3 Views
Open in Slack
PreviousNext
Powered by