https://flink.apache.org/ logo
Join Slack
Powered by
I'm seeing side outputs go to my main datastream s...
# troubleshooting
n
I'm seeing side outputs go to my main datastream sink. I did 
datastream.add_sink(<one test sink>)
and then 
datastream.get_side_output(<output tag>).add_sink(<another test sink>)
. However, my side outputs are showing up in the first sink. I followed the example in https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/side_output/ which seemed quite simple, so I'm not sure what I could do wrong.
Copy code
class InputBroadcastProcessFunctionTests(
        test_case_utils.PyFlinkStreamingTestCase):
    def setUp(self) -> None:
        self.env.set_parallelism(1)
        self.sentinel_sink = testing_utils.DataStreamTestSinkFunction()
        self.timestamp_sink = testing_utils.DataStreamTestSinkFunction()

    def _build_and_run_job(self, datastream: datastream.DataStream,
                           broadcast_stream: datastream.DataStream) -> None:
        processor = input_process_function.InputEventProcessFunction()
        process_results = datastream.connect(
            broadcast_stream.broadcast(
                processor.SENTINEL_BY_ID_DESC)).process(processor)

        process_results.add_sink(self.sentinel_sink)
        process_results.get_side_output(
            sentinel_output.SOURCE_TIME_SIDE_OUTPUT).add_sink(
                self.timestamp_sink)

        self.env.execute('test_input_broadcast_function')
Here's the relevant code if it's of interest.
Oh, maybe this just isn't supported from a 
BroadcastProcessFunction
That doesn't seem true of the Java side since I've seen examples of generating side outputs from a broadcast process function.
It really seems like it's just yielding a tuple instead of a pickled object to a side output
Tracked it so far to see that https://github.com/apache/flink/blob/master/flink-python/pyflink/fn_execution/datastream/process/operations.py#L156 is returning False. I see no mentions in the docs about how this is generated.
Okay it must just be not supported in the python layer. I can make a simple process function to get around it, but sad that the broadcast function can't emit side outputs. Really looks like everything is there for it.
d
cc @胡俊涛(Juntao Hu)
u
Hi Nathanael, it seems an issue when preprocessing python side-out before generating job graph, the broadcast operator itself already supports side-output . I’ll make a quick fix on it.
n
Id be very interested to see where that happens. I only had so much luck trying to reverse where that setting was being applied.
u
You could check out PythonConfigUtil#processSideOutput, where we traverse through the graph to mark side-outputs on python operators.
3 Views
Open in Slack
PreviousNext
Powered by