Re: Issue with growing state/checkpoint size
Hi, Sorry for the late response - was busy with pretty similar problem. First let me clarify on the pipeline You use first. You have a pipeline with a step produces two stream outputs. They do flow thru some computation independently and then You join them by CoGroupByKey with session window and expect to have match in message id. So You are trying to achieve streaming inner join (which should emit output when same message id arrives on both streams). Is it right? My finding: 1. sessions windows are very tricky to work with CoGroupByKey as they are merging windows (so beam may need to wait arbitrary long accumulating state if one of the streams continuously outputs same message id - this is first thing to check in You case). This is only one of potential corner cases with session windows. But I would highly advice to change this to just plain sliding window on computation time domain as it would allow You to clearly specify how long step should keep waiting for particular message id and how frequently it should emit. 2. There is also a case that one of the streams might not produce a message id for particular message. Double make sure that every message that is processes by You pipelines emits only one message per message id always. You also need to consider what to do with the state once pair of messages is joined and emitted. By default SessionWindows will wait for next messages with same message id accumulating state for some time after. This also might be the source of Your problem. Now the solution that I worked for me in similar problem. 1. window both streams in FixedWidnows(1) on processing time domain. Make sure keys are set. 2. CoGroupByKey 3. window output of CoGroupByKey into GlobalWindows 4. Create state-full ParDoFn with timers (to do cleanup) that captures state if a message arrives on one stream, emits both if both messages arrived, does proper action (up to You) when next messages with same key arrive) and does the cleanup. Explanation; FixedWidnows - make sure that elements are groupped in 1s batches to asses join - this is just to meet requirement from CoGroupByKey to have windows defined on input streams. CoGroupByKey - does join messages that arrive on same second window GlobalWindows - makes sure that state-full ParDoFn keeps state globally per key (as it will manage its state by itself). state-full ParDoFn - does the waiting part for the message and contains all Your logic for the corner cases defined above. above just worked perforce for me. Let me know If its a solution for Your case too. We might want to introduce some common api for such cases (its quite easy to code this in generic form). Best Wiśniowski Piotr On 1.09.2023 15:56, Byron Ellis via user wrote: Depends on why you're using a fan-out approach in the first place. You might actually be better off doing all the work at the same time. On Fri, Sep 1, 2023 at 6:43 AM Ruben Vargas wrote: Ohh I see That makes sense. Wondering if there is an strategy for my use case, where I have an ID unique per pair of messages Thanks for all your help! On Fri, Sep 1, 2023 at 6:51 AM Sachin Mittal wrote: Yes a very high and non deterministic cardinality can make the stored state of join operation unbounded. In my case we know the cardinality and it was not very high so we could go with a lookup based approach using redis to enrich the stream and avoid joins. On Wed, Aug 30, 2023 at 5:04 AM Ruben Vargas wrote: Thanks for the reply and the advice One more thing, Do you know if the key-space carnality impacts on this? I'm assuming it is, but the thing is for my case all the messages from the sources has a unique ID, that makes my key-space huge and is not on my control . On Tue, Aug 29, 2023 at 9:29 AM Sachin Mittal wrote: So for the smaller size of collection which does not grow with size for certain keys we stored the data in redis and instead of beam join in our DoFn we just did the lookup and got the data we need. On Tue, 29 Aug 2023 at 8:50 PM, Ruben Vargas wrote: Hello, Thanks for the reply, Any strategy you followed to avoid joins when you rewrite your pipeline? On Tue, Aug 29, 2023 at 9:15 AM Sachin Mittal wrote: Yes even we faced the same issue when trying to run a pipeline involving join of two collections. It was deployed using AWS KDA, which uses flink runner. The source was kinesis streams. Looks like join operations are not very
Re: Issue with growing state/checkpoint size
Depends on why you're using a fan-out approach in the first place. You might actually be better off doing all the work at the same time. On Fri, Sep 1, 2023 at 6:43 AM Ruben Vargas wrote: > Ohh I see > > That makes sense. Wondering if there is an strategy for my use case, where > I have an ID unique per pair of messages > > Thanks for all your help! > > On Fri, Sep 1, 2023 at 6:51 AM Sachin Mittal wrote: > >> Yes a very high and non deterministic cardinality can make the stored >> state of join operation unbounded. >> In my case we know the cardinality and it was not very high so we could >> go with a lookup based approach using redis to enrich the stream and avoid >> joins. >> >> >> >> On Wed, Aug 30, 2023 at 5:04 AM Ruben Vargas >> wrote: >> >>> Thanks for the reply and the advice >>> >>> One more thing, Do you know if the key-space carnality impacts on this? >>> I'm assuming it is, but the thing is for my case all the messages from the >>> sources has a unique ID, that makes my key-space huge and is not on my >>> control . >>> >>> On Tue, Aug 29, 2023 at 9:29 AM Sachin Mittal >>> wrote: >>> So for the smaller size of collection which does not grow with size for certain keys we stored the data in redis and instead of beam join in our DoFn we just did the lookup and got the data we need. On Tue, 29 Aug 2023 at 8:50 PM, Ruben Vargas wrote: > Hello, > > Thanks for the reply, Any strategy you followed to avoid joins when > you rewrite your pipeline? > > > > On Tue, Aug 29, 2023 at 9:15 AM Sachin Mittal > wrote: > >> Yes even we faced the same issue when trying to run a pipeline >> involving join of two collections. It was deployed using AWS KDA, which >> uses flink runner. The source was kinesis streams. >> >> Looks like join operations are not very efficient in terms of size >> management when run on flink. >> >> We had to rewrite our pipeline to avoid these joins. >> >> Thanks >> Sachin >> >> >> On Tue, 29 Aug 2023 at 7:00 PM, Ruben Vargas >> wrote: >> >>> Hello >>> >>> I experimenting an issue with my beam pipeline >>> >>> I have a pipeline in which I split the work into different branches, >>> then I do a join using CoGroupByKey, each message has its own unique >>> Key. >>> >>> For the Join, I used a Session Window, and discarding the messages >>> after trigger. >>> >>> I'm using Flink Runner and deployed a KInesis application. But I'm >>> experiencing an unbounded growth of the checkpoint data size. When I >>> see >>> in Flink console, the following task has the largest checkpoint >>> >>> join_results/GBK -> ToGBKResult -> >>> join_results/ConstructCoGbkResultFn/ParMultiDo(ConstructCoGbkResult) -> >>> V >>> >>> >>> Any Advice ? >>> >>> Thank you very much! >>> >>>
Re: Issue with growing state/checkpoint size
Ohh I see That makes sense. Wondering if there is an strategy for my use case, where I have an ID unique per pair of messages Thanks for all your help! On Fri, Sep 1, 2023 at 6:51 AM Sachin Mittal wrote: > Yes a very high and non deterministic cardinality can make the stored > state of join operation unbounded. > In my case we know the cardinality and it was not very high so we could go > with a lookup based approach using redis to enrich the stream and avoid > joins. > > > > On Wed, Aug 30, 2023 at 5:04 AM Ruben Vargas > wrote: > >> Thanks for the reply and the advice >> >> One more thing, Do you know if the key-space carnality impacts on this? >> I'm assuming it is, but the thing is for my case all the messages from the >> sources has a unique ID, that makes my key-space huge and is not on my >> control . >> >> On Tue, Aug 29, 2023 at 9:29 AM Sachin Mittal wrote: >> >>> So for the smaller size of collection which does not grow with size for >>> certain keys we stored the data in redis and instead of beam join in our >>> DoFn we just did the lookup and got the data we need. >>> >>> >>> On Tue, 29 Aug 2023 at 8:50 PM, Ruben Vargas >>> wrote: >>> Hello, Thanks for the reply, Any strategy you followed to avoid joins when you rewrite your pipeline? On Tue, Aug 29, 2023 at 9:15 AM Sachin Mittal wrote: > Yes even we faced the same issue when trying to run a pipeline > involving join of two collections. It was deployed using AWS KDA, which > uses flink runner. The source was kinesis streams. > > Looks like join operations are not very efficient in terms of size > management when run on flink. > > We had to rewrite our pipeline to avoid these joins. > > Thanks > Sachin > > > On Tue, 29 Aug 2023 at 7:00 PM, Ruben Vargas > wrote: > >> Hello >> >> I experimenting an issue with my beam pipeline >> >> I have a pipeline in which I split the work into different branches, >> then I do a join using CoGroupByKey, each message has its own unique Key. >> >> For the Join, I used a Session Window, and discarding the messages >> after trigger. >> >> I'm using Flink Runner and deployed a KInesis application. But I'm >> experiencing an unbounded growth of the checkpoint data size. When I see >> in Flink console, the following task has the largest checkpoint >> >> join_results/GBK -> ToGBKResult -> >> join_results/ConstructCoGbkResultFn/ParMultiDo(ConstructCoGbkResult) -> V >> >> >> Any Advice ? >> >> Thank you very much! >> >>
Re: Issue with growing state/checkpoint size
Yes a very high and non deterministic cardinality can make the stored state of join operation unbounded. In my case we know the cardinality and it was not very high so we could go with a lookup based approach using redis to enrich the stream and avoid joins. On Wed, Aug 30, 2023 at 5:04 AM Ruben Vargas wrote: > Thanks for the reply and the advice > > One more thing, Do you know if the key-space carnality impacts on this? > I'm assuming it is, but the thing is for my case all the messages from the > sources has a unique ID, that makes my key-space huge and is not on my > control . > > On Tue, Aug 29, 2023 at 9:29 AM Sachin Mittal wrote: > >> So for the smaller size of collection which does not grow with size for >> certain keys we stored the data in redis and instead of beam join in our >> DoFn we just did the lookup and got the data we need. >> >> >> On Tue, 29 Aug 2023 at 8:50 PM, Ruben Vargas >> wrote: >> >>> Hello, >>> >>> Thanks for the reply, Any strategy you followed to avoid joins when you >>> rewrite your pipeline? >>> >>> >>> >>> On Tue, Aug 29, 2023 at 9:15 AM Sachin Mittal >>> wrote: >>> Yes even we faced the same issue when trying to run a pipeline involving join of two collections. It was deployed using AWS KDA, which uses flink runner. The source was kinesis streams. Looks like join operations are not very efficient in terms of size management when run on flink. We had to rewrite our pipeline to avoid these joins. Thanks Sachin On Tue, 29 Aug 2023 at 7:00 PM, Ruben Vargas wrote: > Hello > > I experimenting an issue with my beam pipeline > > I have a pipeline in which I split the work into different branches, > then I do a join using CoGroupByKey, each message has its own unique Key. > > For the Join, I used a Session Window, and discarding the messages > after trigger. > > I'm using Flink Runner and deployed a KInesis application. But I'm > experiencing an unbounded growth of the checkpoint data size. When I see > in Flink console, the following task has the largest checkpoint > > join_results/GBK -> ToGBKResult -> > join_results/ConstructCoGbkResultFn/ParMultiDo(ConstructCoGbkResult) -> V > > > Any Advice ? > > Thank you very much! > >
Re: Issue with growing state/checkpoint size
Thanks for the reply and the advice One more thing, Do you know if the key-space carnality impacts on this? I'm assuming it is, but the thing is for my case all the messages from the sources has a unique ID, that makes my key-space huge and is not on my control . On Tue, Aug 29, 2023 at 9:29 AM Sachin Mittal wrote: > So for the smaller size of collection which does not grow with size for > certain keys we stored the data in redis and instead of beam join in our > DoFn we just did the lookup and got the data we need. > > > On Tue, 29 Aug 2023 at 8:50 PM, Ruben Vargas > wrote: > >> Hello, >> >> Thanks for the reply, Any strategy you followed to avoid joins when you >> rewrite your pipeline? >> >> >> >> On Tue, Aug 29, 2023 at 9:15 AM Sachin Mittal wrote: >> >>> Yes even we faced the same issue when trying to run a pipeline involving >>> join of two collections. It was deployed using AWS KDA, which uses flink >>> runner. The source was kinesis streams. >>> >>> Looks like join operations are not very efficient in terms of size >>> management when run on flink. >>> >>> We had to rewrite our pipeline to avoid these joins. >>> >>> Thanks >>> Sachin >>> >>> >>> On Tue, 29 Aug 2023 at 7:00 PM, Ruben Vargas >>> wrote: >>> Hello I experimenting an issue with my beam pipeline I have a pipeline in which I split the work into different branches, then I do a join using CoGroupByKey, each message has its own unique Key. For the Join, I used a Session Window, and discarding the messages after trigger. I'm using Flink Runner and deployed a KInesis application. But I'm experiencing an unbounded growth of the checkpoint data size. When I see in Flink console, the following task has the largest checkpoint join_results/GBK -> ToGBKResult -> join_results/ConstructCoGbkResultFn/ParMultiDo(ConstructCoGbkResult) -> V Any Advice ? Thank you very much!
Re: Issue with growing state/checkpoint size
So for the smaller size of collection which does not grow with size for certain keys we stored the data in redis and instead of beam join in our DoFn we just did the lookup and got the data we need. On Tue, 29 Aug 2023 at 8:50 PM, Ruben Vargas wrote: > Hello, > > Thanks for the reply, Any strategy you followed to avoid joins when you > rewrite your pipeline? > > > > On Tue, Aug 29, 2023 at 9:15 AM Sachin Mittal wrote: > >> Yes even we faced the same issue when trying to run a pipeline involving >> join of two collections. It was deployed using AWS KDA, which uses flink >> runner. The source was kinesis streams. >> >> Looks like join operations are not very efficient in terms of size >> management when run on flink. >> >> We had to rewrite our pipeline to avoid these joins. >> >> Thanks >> Sachin >> >> >> On Tue, 29 Aug 2023 at 7:00 PM, Ruben Vargas >> wrote: >> >>> Hello >>> >>> I experimenting an issue with my beam pipeline >>> >>> I have a pipeline in which I split the work into different branches, >>> then I do a join using CoGroupByKey, each message has its own unique Key. >>> >>> For the Join, I used a Session Window, and discarding the messages after >>> trigger. >>> >>> I'm using Flink Runner and deployed a KInesis application. But I'm >>> experiencing an unbounded growth of the checkpoint data size. When I see >>> in Flink console, the following task has the largest checkpoint >>> >>> join_results/GBK -> ToGBKResult -> >>> join_results/ConstructCoGbkResultFn/ParMultiDo(ConstructCoGbkResult) -> V >>> >>> >>> Any Advice ? >>> >>> Thank you very much! >>> >>>
Re: Issue with growing state/checkpoint size
Hello, Thanks for the reply, Any strategy you followed to avoid joins when you rewrite your pipeline? On Tue, Aug 29, 2023 at 9:15 AM Sachin Mittal wrote: > Yes even we faced the same issue when trying to run a pipeline involving > join of two collections. It was deployed using AWS KDA, which uses flink > runner. The source was kinesis streams. > > Looks like join operations are not very efficient in terms of size > management when run on flink. > > We had to rewrite our pipeline to avoid these joins. > > Thanks > Sachin > > > On Tue, 29 Aug 2023 at 7:00 PM, Ruben Vargas > wrote: > >> Hello >> >> I experimenting an issue with my beam pipeline >> >> I have a pipeline in which I split the work into different branches, then >> I do a join using CoGroupByKey, each message has its own unique Key. >> >> For the Join, I used a Session Window, and discarding the messages after >> trigger. >> >> I'm using Flink Runner and deployed a KInesis application. But I'm >> experiencing an unbounded growth of the checkpoint data size. When I see >> in Flink console, the following task has the largest checkpoint >> >> join_results/GBK -> ToGBKResult -> >> join_results/ConstructCoGbkResultFn/ParMultiDo(ConstructCoGbkResult) -> V >> >> >> Any Advice ? >> >> Thank you very much! >> >>
Re: Issue with growing state/checkpoint size
Yes even we faced the same issue when trying to run a pipeline involving join of two collections. It was deployed using AWS KDA, which uses flink runner. The source was kinesis streams. Looks like join operations are not very efficient in terms of size management when run on flink. We had to rewrite our pipeline to avoid these joins. Thanks Sachin On Tue, 29 Aug 2023 at 7:00 PM, Ruben Vargas wrote: > Hello > > I experimenting an issue with my beam pipeline > > I have a pipeline in which I split the work into different branches, then > I do a join using CoGroupByKey, each message has its own unique Key. > > For the Join, I used a Session Window, and discarding the messages after > trigger. > > I'm using Flink Runner and deployed a KInesis application. But I'm > experiencing an unbounded growth of the checkpoint data size. When I see > in Flink console, the following task has the largest checkpoint > > join_results/GBK -> ToGBKResult -> > join_results/ConstructCoGbkResultFn/ParMultiDo(ConstructCoGbkResult) -> V > > > Any Advice ? > > Thank you very much! > >
Issue with growing state/checkpoint size
Hello I experimenting an issue with my beam pipeline I have a pipeline in which I split the work into different branches, then I do a join using CoGroupByKey, each message has its own unique Key. For the Join, I used a Session Window, and discarding the messages after trigger. I'm using Flink Runner and deployed a KInesis application. But I'm experiencing an unbounded growth of the checkpoint data size. When I see in Flink console, the following task has the largest checkpoint join_results/GBK -> ToGBKResult -> join_results/ConstructCoGbkResultFn/ParMultiDo(ConstructCoGbkResult) -> V Any Advice ? Thank you very much!