Starting with just checking some basic things: What is the space of keys? Is it constantly growing? Are you explicitly clearing out state from stale keys? In the global window, you don't get any state GC for free.
Can you share repro code? Kenn On Fri, Jan 17, 2020 at 8:53 AM Stephen Patel <stephenpate...@gmail.com> wrote: > I've got a beam pipeline using the FlinkRunner that reads from two > different SQS sources (using the SqsIO). It does some stateful > processing on each stream, and then cogroups the results together to > generate a result and write it to Sns (using the SnsIO). The volume > of input data isn't particularly large (about 50 messages per minute > on one queue, and about 1 message per minute on the other queue). > It's using the Global window, discarding fired panes, with a > processing time trigger delayed by 1 minute. Checkpointing is enabled > at a 1 minute interval, with a minimum delay between checkpoints of 30 > seconds. My state backend is RocksDB, using the FLASH_SSD_OPTIMIZED > predefined options. > > This pipeline runs fine for a few hours with an average checkpoint > duration of 1s (with occasional spikes higher), but eventually the > time it takes to checkpoint begins to grow until it's in the minutes > on average, and finally it won't even complete within a 10 minute > period. I'm using 2 parallelism, and it seems to keep up with the > number of incoming messages just fine (until the checkpoint duration > grows too large and it is unable to delete the messages any longer). > To try to isolate the problem, I wrote an alternate sqs reader that > uses the Watch transform to periodically read from SQS. This variant > doesn't show the same behavior, and has been running for a week > without issue (an average checkpoint time of 1-2s). > > Some other experiments I tried: > > * I observed that the operators that took a long time to checkpoint > were the deduplicating operators after the actual unbounded source > operator. I disabled requiresDeduping and added a Reshuffle instead, > however that exhibited the same growth in checkpoint durations after a > period of time. > * I tried with the AT_LEAST_ONCE checkpointing mode instead of exactly > once, however that also exhibited the same behavior. > > > Does anyone have any thoughts about what might cause this behavior > with the Unbounded Source (as opposed to the splittable do variant)? > > I'm running on EMR emr-5.26.0, using Flink 1.8.0, and Beam 2.14.0. > > Stephen >
