OK, I see your point. I think the demand for updating states and emitting outputs upon receiving a broadcast record makes sense. However, the way `KeyedBroadcastProcessFunction` supports this may not be optimal. E.g., if `Collector#collect` is called in `processBroadcastElement` but outside of `Context#applyToKeyedState`, the result can be undefined.
Currently in this FLIP, a `TwoInputStreamProcessFunction` is not aware of which input is KeyedStream and which is BroadcastStream, which makes supporting things like `applyToKeyedState` difficult. I think we can provide a built-in function similar to `KeyedBroadcastProcessFunction` on top of `TwoInputStreamProcessFunction` to address this demand. WDYT? Best, Xintong On Thu, Feb 1, 2024 at 10:41 AM Xuannan Su <suxuanna...@gmail.com> wrote: > Hi Weijie and Xingtong, > > Thanks for the reply! Please see my comments below. > > > Does this mean if we want to support (KeyedStream, BroadcastStream) -> > > (KeyedStream), we must make sure that no data can be output upon > processing > > records from the input BroadcastStream? That's probably a reasonable > > limitation. > > I don't think that the requirement for supporting (KeyedStream, > BroadcastStream) -> (KeyedStream) is that no data can be output upon > processing the BroadcastStream. For instance, in the current > `KeyedBroadcastProcessFunction`, we use Context#applyToKeyedState to > produce output results, which can be keyed in the same manner as the > keyed input stream, upon processing data from the BroadcastStream. > Therefore, I believe it only requires that the user must ensure that > the output is keyed in the same way as the input, in this case, the > same way as the keyed input stream. I think this requirement is > consistent with that of (KeyedStream, KeyedStream) -> (KeyedStream). > Thus, I believe that supporting (KeyedStream, BroadcastStream) -> > (KeyedStream) will not introduce complexity for the users. WDYT? > > Best regards, > Xuannan > > > On Tue, Jan 30, 2024 at 3:12 PM weijie guo <guoweijieres...@gmail.com> > wrote: > > > > Hi Xintong, > > > > Thanks for your reply. > > > > > Does this mean if we want to support (KeyedStream, BroadcastStream) -> > > (KeyedStream), we must make sure that no data can be output upon > processing > > records from the input BroadcastStream? That's probably a reasonable > > limitation. > > > > I think so, this is the restriction that has to be imposed in order to > > avoid re-partition(i.e. shuffle). > > If one just want to get a keyed-stream and don't care about the data > > distribution, then explicit KeyBy partitioning works as expected. > > > > > The problem is would this limitation be too implicit for the users to > > understand. > > > > Since we can't check for this limitation at compile time, if we were to > add > > support for this case, we would have to introduce additional runtime > checks > > to ensure program correctness. For now, I'm inclined not to support it, > as > > it's hard for users to understand this restriction unless we have > something > > better. And we can always add it later if we do realize there's a strong > > demand for it. > > > > > 1. I'd suggest renaming the method with timestamp to something like > > `collectAndOverwriteTimestamp`. That might help users understand that > they > > don't always need to call this method, unless they explicitly want to > > overwrite the timestamp. > > > > Make sense, I have updated this FLIP toward this new method name. > > > > > 2. While this method provides a way to set timestamps, how would users > > read > > timestamps from the records? > > > > Ah, good point. I will introduce a new method to get the timestamp of the > > current record in RuntimeContext. > > > > > > Best regards, > > > > Weijie > > > > > > Xintong Song <tonysong...@gmail.com> 于2024年1月30日周二 14:04写道： > > > > > Just trying to understand. > > > > > > > Is there a particular reason we do not support a > > > > `TwoInputProcessFunction` to combine a KeyedStream with a > > > > BroadcastStream to result in a KeyedStream? There seems to be a valid > > > > use case where a KeyedStream is enriched with a BroadcastStream and > > > > returns a Stream that is partitioned in the same way. > > > > > > > > > > The key point here is that if the returned stream is a KeyedStream, > we > > > > require that the partition of input and output be the same. As for > the > > > > data on the broadcast edge, it will be broadcast to all parallelism, > we > > > > cannot keep the data partition consistent. For example, if a specific > > > > record is sent to both SubTask1 and SubTask2, after processing, the > > > > partition index calculated by the new KeySelector is `1`, then the > data > > > > distribution of SubTask2 has obviously changed. > > > > > > > > > Does this mean if we want to support (KeyedStream, BroadcastStream) -> > > > (KeyedStream), we must make sure that no data can be output upon > processing > > > records from the input BroadcastStream? That's probably a reasonable > > > limitation. The problem is would this limitation be too implicit for > the > > > users to understand. > > > > > > > > > > I noticed that there are two `collect` methods in the Collector, > > > > > > > > one with a timestamp and one without. Could you elaborate on the > > > > differences between them? Additionally, in what use case would one > use > > > > the method that includes the timestamp? > > > > > > > > > > > > That's a good question, and it's mostly used with time-related > operators > > > > such as Window. First, we want to give the process function the > ability > > > to > > > > reset timestamps, which makes it more flexible than the original > > > > API. Second, we don't want to take the timestamp extraction > > > > operator/function as a base primitive, it's more like a high-level > > > > extension. Therefore, the framework must provide this functionality. > > > > > > > > > > > 1. I'd suggest renaming the method with timestamp to something like > > > `collectAndOverwriteTimestamp`. That might help users understand that > they > > > don't always need to call this method, unless they explicitly want to > > > overwrite the timestamp. > > > > > > 2. While this method provides a way to set timestamps, how would users > read > > > timestamps from the records? > > > > > > > > > Best, > > > > > > Xintong > > > > > > > > > > > > On Tue, Jan 30, 2024 at 12:45 PM weijie guo <guoweijieres...@gmail.com > > > > > wrote: > > > > > > > Hi Xuannan, > > > > > > > > Thank you for your attention. > > > > > > > > > In the partitioning section, it says that "broadcast can only be > > > > used as a side-input of other Inputs." Could you clarify what is > meant > > > > by "side-input"? If I understand correctly, it refer to one of the > > > > inputs of the `TwoInputStreamProcessFunction`. If that is the case, > > > > the term "side-input" may not be accurate. > > > > > > > > Yes, you got it right! I have rewrote this sentence to avoid > > > > misunderstanding. > > > > > > > > > Is there a particular reason we do not support a > > > > `TwoInputProcessFunction` to combine a KeyedStream with a > > > > BroadcastStream to result in a KeyedStream? There seems to be a valid > > > > use case where a KeyedStream is enriched with a BroadcastStream and > > > > returns a Stream that is partitioned in the same way. > > > > > > > > The key point here is that if the returned stream is a KeyedStream, > we > > > > require that the partition of input and output be the same. As for > the > > > > data on the broadcast edge, it will be broadcast to all parallelism, > we > > > > cannot keep the data partition consistent. For example, if a specific > > > > record is sent to both SubTask1 and SubTask2, after processing, the > > > > partition index calculated by the new KeySelector is `1`, then the > data > > > > distribution of SubTask2 has obviously changed. > > > > > > > > > 3. There appears to be a typo in the example code. The > > > > `SingleStreamProcessFunction` should probably be > > > > `OneInputStreamProcessFunction`. > > > > > > > > Yes, good catch. I have updated this FLIP. > > > > > > > > > 4. How do we set the global configuration for the > > > > ExecutionEnvironment? Currently, we have the > > > > StreamExecutionEnvironment.getExecutionEnvironment(Configuration) > > > > method to provide the global configuration in the API. > > > > > > > > This is because we don't want to allow set config programmatically > in the > > > > new API, everything best comes from configuration files. However, > this > > > may > > > > be too ideal, and the specific details need to be considered and > > > discussed > > > > in more detail, and I propose to devote a new sub-FLIP to this issue > > > later. > > > > We can easily provide the `getExecutionEnvironment(Configuration)` or > > > > `withConfiguration(Configuration)` method later. > > > > > > > > > I noticed that there are two `collect` methods in the Collector, > > > > one with a timestamp and one without. Could you elaborate on the > > > > differences between them? Additionally, in what use case would one > use > > > > the method that includes the timestamp? > > > > > > > > That's a good question, and it's mostly used with time-related > operators > > > > such as Window. First, we want to give the process function the > ability > > > to > > > > reset timestamps, which makes it more flexible than the original > > > > API. Second, we don't want to take the timestamp extraction > > > > operator/function as a base primitive, it's more like a high-level > > > > extension. Therefore, the framework must provide this functionality. > > > > > > > > > > > > Best regards, > > > > > > > > Weijie > > > > > > > > > > > > weijie guo <guoweijieres...@gmail.com> 于2024年1月30日周二 11:45写道： > > > > > > > > > Hi Yunfeng, > > > > > > > > > > Thank you for your attention > > > > > > > > > > > 1. Will we provide any API to support choosing which input to > consume > > > > > between the two inputs of TwoInputStreamProcessFunction? It would > be > > > > > helpful in online machine learning cases, where a process function > > > > > needs to receive the first machine learning model before it can > start > > > > > predictions on input data. Similar requirements might also exist in > > > > > Flink CEP, where a rule set needs to be consumed by the process > > > > > function first before it can start matching the event stream > against > > > > > CEP patterns. > > > > > > > > > > Good point! I think we can provide a `nextInputSelection()` method > for > > > > > `TwoInputStreamProcessFunction`. It returns a ·First/Second· enum > that > > > > > determines which Input the mailbox thread will read next. But I'm > > > > > considering putting it in the sub-FLIP related to Join, since > features > > > > like > > > > > HashJoin have a more specific need for this. > > > > > > > > > > > A typo might exist in the current FLIP describing the API to > > > > > generate a global stream, as I can see either global() or > coalesce() > > > > > in different places of the FLIP. These two methods might need to be > > > > > unified into one method. > > > > > > > > > > Good catch! I have updated this FLIP to fix this typo. > > > > > > > > > > > The order of parameters in the current ProcessFunction is > (record, > > > > > context, output), while this FLIP proposes to change the order into > > > > > (record, output, context). Is there any reason to make this change? > > > > > > > > > > No, it's just the order we decide. But please note that there is no > > > > > relationship between the two ProcessFunction's anyway. I think it's > > > okay > > > > to > > > > > use our own order of parameters in new API. > > > > > > > > > > 4. Why does this FLIP propose to use connectAndProcess() instead of > > > > > connect() (+ keyBy()) + process()? The latter looks simpler to me. > > > > > > > > > > > I actually also considered this way at first, but it would have > to > > > > > introduce some concepts like ConnectedStreams. But we hope that > streams > > > > > will be more clearly defined in the DataStream API, otherwise we > will > > > end > > > > > up going the same way as the original API, which you have to > understand > > > > > `JoinedStreams/ConnectedStreams` and so on. > > > > > > > > > > > > > > > > > > > > Best regards, > > > > > > > > > > Weijie > > > > > > > > > > > > > > > weijie guo <guoweijieres...@gmail.com> 于2024年1月30日周二 11:20写道： > > > > > > > > > >> Hi Wencong: > > > > >> > > > > >> Thank you for your attention > > > > >> > > > > >> > Q1. Other DataStream types are converted into > > > > >> Non-Keyed DataStreams by using a "shuffle" operation > > > > >> to convert Input into output. Does this "shuffle" include the > > > > >> various repartition operations (rebalance/rescale/shuffle) > > > > >> from DataStream V1? > > > > >> > > > > >> Yes, The name `shuffle` is used only to represent the > transformation > > > of > > > > >> an arbitrary stream into a non-keyed partitioned stream and does > not > > > > >> restrict how the data is partitioned. > > > > >> > > > > >> > > > > >> > Q2. Why is the design for TwoOutputStreamProcessFunction, > > > > >> when dealing with a KeyedStream, only outputting combinations > > > > >> of (Keyed + Keyed) and (Non-Keyed + Non-Keyed)? > > > > >> > > > > >> In theory, we could only provide functions that return Non-Keyed > > > > streams. > > > > >> If you do want a KeyedStream, you explicitly convert it to a > > > KeyedStream > > > > >> via keyBy. However, because sometimes data is processed without > > > changing > > > > >> the partition, we choose to provide an additional KeyedStream > > > > counterpart > > > > >> to reduce the shuffle overhead. We didn't introduce the non-keyed > + > > > > >> keyed combo here simply because it's not very common, and if we > really > > > > see > > > > >> a lot of users asking for it later on, it's easy to support it > then. > > > > >> > > > > >> > > > > >> Best regards, > > > > >> > > > > >> Weijie > > > > >> > > > > >> > > > > >> Xuannan Su <suxuanna...@gmail.com> 于2024年1月29日周一 18:28写道： > > > > >> > > > > >>> Hi Weijie, > > > > >>> > > > > >>> Thank you for driving the design of the new DataStream API. I > have a > > > > >>> few questions regarding the FLIP: > > > > >>> > > > > >>> 1. In the partitioning section, it says that "broadcast can only > be > > > > >>> used as a side-input of other Inputs." Could you clarify what is > > > meant > > > > >>> by "side-input"? If I understand correctly, it refer to one of > the > > > > >>> inputs of the `TwoInputStreamProcessFunction`. If that is the > case, > > > > >>> the term "side-input" may not be accurate. > > > > >>> > > > > >>> 2. Is there a particular reason we do not support a > > > > >>> `TwoInputProcessFunction` to combine a KeyedStream with a > > > > >>> BroadcastStream to result in a KeyedStream? There seems to be a > valid > > > > >>> use case where a KeyedStream is enriched with a BroadcastStream > and > > > > >>> returns a Stream that is partitioned in the same way. > > > > >>> > > > > >>> 3. There appears to be a typo in the example code. The > > > > >>> `SingleStreamProcessFunction` should probably be > > > > >>> `OneInputStreamProcessFunction`. > > > > >>> > > > > >>> 4. How do we set the global configuration for the > > > > >>> ExecutionEnvironment? Currently, we have the > > > > >>> StreamExecutionEnvironment.getExecutionEnvironment(Configuration) > > > > >>> method to provide the global configuration in the API. > > > > >>> > > > > >>> 5. I noticed that there are two `collect` methods in the > Collector, > > > > >>> one with a timestamp and one without. Could you elaborate on the > > > > >>> differences between them? Additionally, in what use case would > one > > > use > > > > >>> the method that includes the timestamp? > > > > >>> > > > > >>> Best regards, > > > > >>> Xuannan > > > > >>> > > > > >>> > > > > >>> > > > > >>> On Fri, Jan 26, 2024 at 2:21 PM Yunfeng Zhou > > > > >>> <flink.zhouyunf...@gmail.com> wrote: > > > > >>> > > > > > >>> > Hi Weijie, > > > > >>> > > > > > >>> > Thanks for raising discussions about the new DataStream API. I > > > have a > > > > >>> > few questions about the content of the FLIP. > > > > >>> > > > > > >>> > 1. Will we provide any API to support choosing which input to > > > consume > > > > >>> > between the two inputs of TwoInputStreamProcessFunction? It > would > > > be > > > > >>> > helpful in online machine learning cases, where a process > function > > > > >>> > needs to receive the first machine learning model before it can > > > start > > > > >>> > predictions on input data. Similar requirements might also > exist in > > > > >>> > Flink CEP, where a rule set needs to be consumed by the process > > > > >>> > function first before it can start matching the event stream > > > against > > > > >>> > CEP patterns. > > > > >>> > > > > > >>> > 2. A typo might exist in the current FLIP describing the API to > > > > >>> > generate a global stream, as I can see either global() or > > > coalesce() > > > > >>> > in different places of the FLIP. These two methods might need > to be > > > > >>> > unified into one method. > > > > >>> > > > > > >>> > 3. The order of parameters in the current ProcessFunction is > > > (record, > > > > >>> > context, output), while this FLIP proposes to change the order > into > > > > >>> > (record, output, context). Is there any reason to make this > change? > > > > >>> > > > > > >>> > 4. Why does this FLIP propose to use connectAndProcess() > instead of > > > > >>> > connect() (+ keyBy()) + process()? The latter looks simpler to > me. > > > > >>> > > > > > >>> > Looking forward to discussing these questions with you. > > > > >>> > > > > > >>> > Best regards, > > > > >>> > Yunfeng Zhou > > > > >>> > > > > > >>> > On Tue, Dec 26, 2023 at 2:44 PM weijie guo < > > > > guoweijieres...@gmail.com> > > > > >>> wrote: > > > > >>> > > > > > > >>> > > Hi devs, > > > > >>> > > > > > > >>> > > > > > > >>> > > I'd like to start a discussion about FLIP-409: DataStream V2 > > > > Building > > > > >>> > > Blocks: DataStream, Partitioning and ProcessFunction [1]. > > > > >>> > > > > > > >>> > > > > > > >>> > > As the first sub-FLIP for DataStream API V2, we'd like to > discuss > > > > and > > > > >>> > > try to answer some of the most fundamental questions in > stream > > > > >>> > > processing: > > > > >>> > > > > > > >>> > > 1. What kinds of data streams do we have? > > > > >>> > > 2. How to partition data over the streams? > > > > >>> > > 3. How to define a processing on the data stream? > > > > >>> > > > > > > >>> > > The answer to these questions involve three core concepts: > > > > >>> DataStream, > > > > >>> > > Partitioning and ProcessFunction. In this FLIP, we will > discuss > > > the > > > > >>> > > definitions and related API primitives of these concepts in > > > detail. > > > > >>> > > > > > > >>> > > > > > > >>> > > You can find more details in FLIP-409 [1]. This sub-FLIP is > at > > > the > > > > >>> > > heart of the entire DataStream API V2, and its relationship > with > > > > >>> other > > > > >>> > > sub-FLIPs can be found in the umbrella FLIP [2]. > > > > >>> > > > > > > >>> > > > > > > >>> > > Looking forward to hearing from you, thanks! > > > > >>> > > > > > > >>> > > > > > > >>> > > Best regards, > > > > >>> > > > > > > >>> > > Weijie > > > > >>> > > > > > > >>> > > > > > > >>> > > > > > > >>> > > [1] > > > > >>> > > > > > > >>> > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-409%3A+DataStream+V2+Building+Blocks%3A+DataStream%2C+Partitioning+and+ProcessFunction > > > > >>> > > > > > > >>> > > [2] > > > > >>> > > > > > > >>> > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-408%3A+%5BUmbrella%5D+Introduce+DataStream+API+V2 > > > > >>> > > > > >> > > > > > > > >
