Processing watermarks in a broadcast connected stream

[Sajjad Rizvi <sa...@arcticwolf.com>]

Hi,

I am trying to process watermarks in a BroadcastConnectedStream. However, I am not able to find any direct way to handle watermark events, similar to what we have in processWatermark1 in a  KeyedCoProcessOperator. Following are further details.

In the context of the example given in “A Practical Guide to Broadcast State in Apache Flink<https://flink.apache.org/2019/06/26/broadcast-state.html>”, I have a user actions stream and a pattern stream. The pattern stream is broadcast and connected with the user actions stream. The result is a BroadcastConnectedStream. I want to handle user action events and pattern evens in this stream. In addition, I want to use a processWatermark function to perform an action in response to watermark events.

The problem is that a BroadcastConnectedStream has only process() function, no transform(), that takes a (Keyed)BroadcastProcessFunction. A BroadcastProcessFunction allows only to process elements, doesn’t provide the interface to process watermarks. In contrast, a ConnectedStream (without broadcast) provides a transform function, which takes in an operator that provides a way to process watermarks.

Is there a way to process watermarks in a BroadcastConnectedStream?

Thanks,
Sajjad

This email and any attachments are for the expressed and sole use of the intended recipient(s) and contain information that may be confidential and/or legally privileged. Any disclosure, copying, distribution or use of this communication by someone other than the intended recipient is strictly prohibited. If you are not the intended recipient please delete this email immediately. Any information and services described herein are provided by Arctic Wolf Networks, Inc.

RE: Processing watermarks in a broadcast connected stream

[Schwalbe Matthias <Ma...@viseca.ch>]

Good Morning Sajjad,

I’ve once had a similar problem. As you’ve found out, directly using KeyedBroadcastProcessFunction is a little tricky.
What I ended up with instead is to use the rather new @PublicEvolving MultipleInputStreamOperator<OUT>.
It allows you to connect and process any (reasonable) number of DataStream keyed/broadcast/plain and also to tap into
the meta-stream of watermark events. Each Input is set up separately and can implement separate handlers for the events/watermarks/etc.
However, it is an operator implementation, you e.g. need to manually set up timer manager and a number of other auxiliary components.
This is not too difficult as you can always model after other operator implementations within flink.

If you don’t mind that it will be in Scala, I could take the time to collect the basic setup …?


Hope this helps

Thias







From: Sajjad Rizvi <sa...@arcticwolf.com>
Sent: Monday, January 30, 2023 7:42 PM
To: user@flink.apache.org
Subject: Processing watermarks in a broadcast connected stream

⚠EXTERNAL MESSAGE – CAUTION: Think Before You Click ⚠


Hi,

I am trying to process watermarks in a BroadcastConnectedStream. However, I am not able to find any direct way to handle watermark events, similar to what we have in processWatermark1 in a  KeyedCoProcessOperator. Following are further details.

In the context of the example given in “A Practical Guide to Broadcast State in Apache Flink<https://flink.apache.org/2019/06/26/broadcast-state.html>”, I have a user actions stream and a pattern stream. The pattern stream is broadcast and connected with the user actions stream. The result is a BroadcastConnectedStream. I want to handle user action events and pattern evens in this stream. In addition, I want to use a processWatermark function to perform an action in response to watermark events.

The problem is that a BroadcastConnectedStream has only process() function, no transform(), that takes a (Keyed)BroadcastProcessFunction. A BroadcastProcessFunction allows only to process elements, doesn’t provide the interface to process watermarks. In contrast, a ConnectedStream (without broadcast) provides a transform function, which takes in an operator that provides a way to process watermarks.

Is there a way to process watermarks in a BroadcastConnectedStream?

Thanks,
Sajjad


Disclaimer

This email and any attachments are for the expressed and sole use of the intended recipient(s) and contain information that may be confidential and/or legally privileged. Any disclosure, copying, distribution or use of this communication by someone other than the intended recipient is strictly prohibited. If you are not the intended recipient please delete this email immediately. Any information and services described herein are provided by Arctic Wolf Networks, Inc.
Diese Nachricht ist ausschliesslich für den Adressaten bestimmt und beinhaltet unter Umständen vertrauliche Mitteilungen. Da die Vertraulichkeit von e-Mail-Nachrichten nicht gewährleistet werden kann, übernehmen wir keine Haftung für die Gewährung der Vertraulichkeit und Unversehrtheit dieser Mitteilung. Bei irrtümlicher Zustellung bitten wir Sie um Benachrichtigung per e-Mail und um Löschung dieser Nachricht sowie eventueller Anhänge. Jegliche unberechtigte Verwendung oder Verbreitung dieser Informationen ist streng verboten.

This message is intended only for the named recipient and may contain confidential or privileged information. As the confidentiality of email communication cannot be guaranteed, we do not accept any responsibility for the confidentiality and the intactness of this message. If you have received it in error, please advise the sender by return e-mail and delete this message and any attachments. Any unauthorised use or dissemination of this information is strictly prohibited.