Hi Xintong, Thank you for all the comments. Please see my reply inline.
On Mon, Sep 18, 2023 at 11:31 AM Xintong Song <tonysong...@gmail.com> wrote: > Thanks for addressing my comments, Dong. > > The expected behavior of checkpointing and failover depends on whether > > there is any operator currently running in the job with all its inputs' > > isBacklog=true. If there exists such an operator and > > interval-during-backlog = 0, then checkpoint will be disabled and the > > operator will have to failover in a way similar to batch mode. > > > This makes sense to me. Shall we also put this into the FLIP. Or maybe you > already did that and I overlooked it? The current description in "4) > Checkpoint and failover strategy" -> "Mixed mode" is a bit unclear to me. > It says "At the point when isBacklog switches to false, source operator > ...", which sounds like upon any source operator switching to isBacklog = > false. > I think it is kind of mentioned in the doc of execution.checkpointing.interval-during-backlog, which says "if it is not null and any source reports isProcessingBacklog=true, it is the interval...". Based on this doc, we can derive that if there is one operator reporting isBacklog=true, then the checkpointing interval is determined by interval-during-backlog, which in this case has value 0 indicating that checkpoint triggering is disabled. Given that other readers might also have this question, I have updated the FLIP-327 with the following statement to make it more explicit: "For jobs with multiple sources and execution.checkpointing.interval-during-backlog = 0, checkpoint triggering is enabled if and only if all sources have isBacklog=false". > I am not sure what is the concern with having `flink-streaming-java` depend > > on `flink-runtime`. Can you clarify the exact concern? > > > > The concern here is that an API module should not depend on a runtime > module. Currently, we have the "user codes -> flink-streaming-java -> > flink-runtime" dependency chain, which makes binary compatibility > impossible because any runtime changes can break the compatibility with a > user jar (which bundles flink-streaming-java) compiled for an older > version. Ideally, we want the runtime module to depend on the API module, > rather than the other way around. This is one of the issues we are trying > to resolve with the programmatic API refactor. However, the way we are > trying to resolve it is to introduce another API module and gradually > replace the current DataStream API / flink-streaming-java, which means > flink-streaming-java will stay depending on flink-runtime for a while > anyway. So the concern here is minor, only about we might need more effort > when reworking this with the new API. > Thanks for the detailed explanation. Given that we plan to avoid having flink-streaming-java depend on flink-runtime, I agree it is preferred to avoid introducing more dependencies like this. I have updated the FLIP to let RecordAttributes extend StreamElement. Best, Dong > The rest of your replies make sense to me. > > Best, > > Xintong > > > > On Fri, Sep 15, 2023 at 10:05 PM Dong Lin <lindon...@gmail.com> wrote: > > > Hi Xintong, > > > > Thanks for your comments! Please see my reply inline. > > > > On Thu, Sep 14, 2023 at 4:58 PM Xintong Song <tonysong...@gmail.com> > > wrote: > > > > > Sorry to join the discussion late. > > > > > > Overall, I think it's a good idea to support dynamically switching the > > > operator algorithms between Streaming (optimized towards low latency + > > > checkpointing supports) and Batch (optimized towards throughput). This > is > > > indeed a big and complex topic, and I really appreciate the previous > > > discussions that narrow the scope of this FLIP down to only considering > > > switching from Batch to Streaming as a first step. > > > > > > I have several questions. > > > > > > 1. The FLIP discusses various behaviors under 4 scenarios: streaming > > mode, > > > batch mode, mixed mode with checkpoint interval > 0, mixed mode with > > > checkpoint interval = 0. IIUC, this is because many batch optimizations > > > cannot be supported together with checkpointing. This justifies that in > > > mixed mode with interval > 0, most behaviors are the same as in > streaming > > > mode. However, mixed mode with checkpoint interval = 0 does not always > > > necessarily mean we should apply such optimization. It is possible that > > in > > > some cases (likely with small data amounts) the cost of such > > optimizations > > > are higher than the benefit. Therefore, I'd suggest decoupling the > > concept > > > of applying these optimizations (i.e., the batch execution phase in the > > > mixed mode) from whether checkpointing is enabled or not. In > particular, > > > I'd suggest removing the scenario "mixed mode with > > > e.c.interval-during-backlog > 0", changing the scenario "mixed mode > with > > > e.c.interval-during-backlog = 0" to simply "mixed mode", and say that > can > > > have different strategies for deciding whether to enable the mixed mode > > and > > > as the first step the strategy is to enable it when > > > e.c.interval-during-backlog = 0. > > > > > > > Thanks for the detailed explanation! > > > > I have updated the "Behavior changes when switching from batch mode to > > stream mode" section with the following changes. > > > > 1) Remove the description of "mixed mode with interval-during-backlog > > 0" > > and add the statement saying that "after this FLIP, the behavior of Flink > > runtime with execution.runtime-mode = streaming AND > > execution.checkpointing.interval-during-backlog > 0, will be same as the > > stream mode prior to this FLIP" > > > > 2) Add the statement saying that "Mixed mode refers to the behavior of > > Flink runtime after this FLIP with execution.runtime-mode = streaming AND > > execution.checkpointing.interval-during-backlog = 0". > > > > 3) Add the statement saying that "It is possible for mixed mode to be > > slower than stream mode, particularly when there is only small amount of > > input records and the overhead of buffering/sorting inputs out-weight its > > benefit. This is similar to how the merge join might be slower than hash > > join. This FLIP focuses on optimizing the Flink throughput when there is > a > > high number of input records. In the future, we might introduce more > > strategies to turn on mix mode in a smart way to avoid performance > > regression". > > > > Would this address your concern? > > > > > > > > > 2. According to the FLIP, before isBacklog = false, the timer service > > only > > > keeps timers for the current key. It also says upon the end of each > key, > > it > > > fires timers of the key up to the last watermark. IIUC, that means not > > all > > > timers are guaranteed to be fired. It is possible that some timers are > > left > > > to be triggered after isBacklog switching to false. If the timer > service > > > only keeps timers for the current key, those not-fired timers may get > > lost > > > when switching to a new key. > > > > > > > Thanks for catching this. You are right that all timers should be fired > as > > long as the corresponding firing condition (either processing-time or > > event-time) is satisfied. > > > > I have updated the "Timer Service" part of the "Behavior changes when > > switching from batch mode to stream mode" section accordingly. Can you > see > > if it addresses your concern? > > > > > > > > > > 3. Is it possible that some sources / operators in the job switch to > > > isBacklog = false, while others are still isBacklog = true? In that > case, > > > what is the expected behavior for checkpointing and failover? > > > > > > > Yes, it is possible. And in this case, Flink runtime will handle this > > operator as if all the operator's inputs have isBacklog=false. In > > particular, Flink runtime will not automatically sort inputs of this > > operator. > > > > I added the following statement in the FLIP to clarify the behavior: "For > > an operator with 2+ inputs, where some inputs have isBacklog=true and > some > > other inputs have isBacklog=false, Flink runtime will handle this > operator > > as if all its inputs have isBacklog=false". > > > > The expected behavior of checkpointing and failover depends on whether > > there is any operator currently running in the job with all its inputs' > > isBacklog=true. If there exists such an operator > > and interval-during-backlog = 0, then checkpoint will be disabled and the > > operator will have to failover in a way similar to batch mode. > > > > > > > > > > > > 4. Do we require RecordAttributes to be properly handled by all > > operators? > > > Or do we consider it as hints that operators may benefit from looking > > into > > > it but should not run into any problems ignoring it? I'm asking > because, > > if > > > they are required to be properly handled, we probably need a way to > > enforce > > > operators to deal with it. `processRecordAttributes(RecordAttributes)` > > > might not be a good fit because we don't know whether the operator has > > > looked into all necessary fields of `RecordAttributes`. > > > > > > > As of this FLIP, we would not require RecordAttributes to be handled any > > operator in order to achieve correctness. So it is more like a hint. More > > specifically, the isBacklog attribute provides a hint for an operator to > > optionally delay the processing of its inputs if doing so can improve its > > throughput. > > > > I think it would be useful to avoid requiring operators to explicitly > > handling attributes contained in RecordAttributes. This is because we > want > > the features added in this FLIP (and future FLIPs) to be backward > > compatible without breaking the correctness of existing jobs. > > > > Suppose we really need to add a record attribute that should be > explicitly > > handled by every operator, I believe we can always find a way to enforce > > this requirement (e.g. fail job compilation with proper error) in the > > future. For example, we can add a method such as > handleXXXRecodAttribute() > > in the operator interface without default implementation. > > > > > > > > > > 5. I wonder if there's any strong reasons behind choosing > `RuntimeEvent` > > > over `StreamElement` for `RecordAttributes` to extend? My concern is > > that, > > > the current approach introduces one more dependency from > > > `flink-streaming-java` (operators that uses `RecordAttributes`) to > > > `flink-runtime` (where `RuntimeEvent` comes from), which seems to be > > > unnecessary. > > > > > > > There is no strong reason to choose `RuntimeEvent` over `StreamElement`. > I > > think the main (and minor) reason for doing so is the simplicity of > > implementation. For example, we don't need to add methods such > > as StreamElement#isRecordAttributes and StreamElement#asRecordAttributes. > > > > I am not sure what is the concern with having `flink-streaming-java` > depend > > on `flink-runtime`. Can you clarify the exact concern? > > > > In any case, I don't have a strong preference between `RuntimeEvent` over > > `StreamElement`. We can update the FLIP to use `StreamElement` as long as > > there is a well-defined non-trivial reason for making this choice. > > > > > > > 6. The FLIP says it leverages state backend optimizations introduced in > > > FLIP-325. Just for clarification, does this mean this FLIP is depending > > on > > > FLIP-325, and probably should not be voted / accepted until FLIP-325 is > > > accepted? > > > > > > > Yes, the proposed implementation of this FLIP depends on FLIP-325. We can > > start voting thread for FLIP-327 after FLIP-325 is accepted. Maybe we can > > continue to discuss this FLIP in the mean time (before FLIP-325 is > > accepted). > > > > Thanks again for the very detailed and helpful review! Looking forward to > > your follow-up comments. > > > > Best, > > Dong > > > > > > > Best, > > > > > > Xintong > > > > > > > > > > > > On Fri, Sep 1, 2023 at 12:48 AM Jan Lukavský <je...@seznam.cz> wrote: > > > > > > > Hi, > > > > > > > > some keywords in this triggered my attention, so sorry for late > jumping > > > > in, but I'd like to comprehend the nature of the proposal. > > > > > > > > I'll try to summarize my understanding: > > > > > > > > The goal of the FLIP is to support automatic switching between > > streaming > > > > and batch processing, leveraging the fact that batch processing is > more > > > > computationally effective. This makes perfect sense. > > > > > > > > Looking at the streaming vs. batch semantics, switching from > streaming > > > > to batch means the following: > > > > > > > > a) generally, watermarks are not propagated in batch, watermark > moves > > > > from -inf to +inf in one step, at the end of batch input, this might > > > > (and probably will) skip many invocations of timers > > > > > > > > b) grouping by key (and window) can be done efficiently, because it > > > > can be done by sort-group and ideally parallelized by window (with > some > > > > caveats) > > > > > > > > The switch also has some conditions, namely: > > > > > > > > i) batch mode does not do checkpoints, inputs must be accessible > > > > repeatedly (forever) > > > > > > > > ii) due to failures in batch mode, inputs might be reprocessed and > > > > thus must be immutable or all sub-results computed in all branches of > > > > the computation (even possibly unaffected by the failure) have to be > > > > discarded and recomputed from scratch > > > > > > > > Obviously, in case of the switch from batch to streaming, the > property > > > > a) has to be modified so the watermark does not move to +inf, but to > > > > min(streaming watermark). Giving these properties, it should be > > possible > > > > to exchange batch and streaming processing without any cooperation > with > > > > the application logic itself. Is my understanding correct? > > > > > > > > If so, there is still one open question to efficiency, though. The > > > > streaming operator _might_ need sorting by timestamp (e.g. processing > > > > time-series data, or even sequential data). In that case simply > > > > switching streaming semantics to batch processing does not yield > > > > efficient processing, because the operator still needs to buffer and > > > > manually sort all the input data (batch data is always unordered). On > > > > the other hand, the batch runner already does sorting (for grouping > by > > > > key), so adding additional sorting criterion is very cheap. In Apache > > > > Beam, we introduced a property of a stateful PTransform (DoFn) called > > > > @RequiresTimeSortedInput [1], which can then be implemented > efficiently > > > > by batch engines. > > > > > > > > Does the FLIP somehow work with conditions i) and ii)? I can imagine > > for > > > > instance that if data is read from say Kafka, then if backlog gets > > > > sufficiently large, then even the batch processing can take > substantial > > > > time and if it fails after long processing, some of the original data > > > > might be already rolled out from Kafka topic. > > > > > > > > In the FLIP there are some proposed changes to sources to emit > metadata > > > > about if the records come from backlog. What is the driving line of > > > > thoughts why this is needed? In my point of view, streaming engines > are > > > > _always_ processing backlog, the only question is "how delayed are > the > > > > currently processed events after HEAD", or more specifically in this > > > > case "how many elements can we expect to process if the source would > > > > immediately stop receiving more data?". This should be configurable > > > > using simple option defining the difference between current > > > > processing-time (JM) and watermark of the source, or am I missing > > > > something? > > > > > > > > Thanks for clarification and all the best, > > > > > > > > Jan > > > > > > > > [1] > > > > > > > > > > > > > > https://beam.apache.org/releases/javadoc/2.50.0/org/apache/beam/sdk/transforms/DoFn.RequiresTimeSortedInput.html > > > > > > > > On 8/31/23 13:17, Xuannan Su wrote: > > > > > Hi all, > > > > > > > > > > I would like to share some updates on FLIP-327. Dong and I have > had a > > > > > series of discussions and have made several refinements to the > FLIP. > > > > > > > > > > The major change to the FLIP is to allow the input of the one-input > > > > > operator to be automatically sorted during backlog processing. When > > > > > combined with the state backend optimization introduced in FLIP-325 > > > [1], > > > > > all the keyed single-input operators can achieve similar > performance > > as > > > > in > > > > > batch mode during backlog processing without any code change to the > > > > > operator. We also implemented a POC[2] and conducted benchmark[3] > > using > > > > the > > > > > KeyedStream#reduce operation. The benchmark results demonstrate the > > > > > performance gains that this FLIP can offer. > > > > > > > > > > I am looking forward to any comments or feedback you may have on > this > > > > FLIP. > > > > > > > > > > Best, > > > > > Xuannan > > > > > > > > > > [1] > > > > > > > > > > > > > > > https://cwiki.apache.org/confluence/display/FLINK/FLIP-325%3A+Introduce+LRU+cache+to+accelerate+state+backend+access > > > > > [2] https://github.com/Sxnan/flink/tree/FLIP-327-demo > > > > > [3] > > > > > > > > > > > > > > > https://github.com/Sxnan/flink/blob/d77d0d3fb268de0a1939944ea4796a112e2d68c0/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/backlog/ReduceBacklogBenchmark.java > > > > > > > > > > > > > > > > > > > >> On Aug 18, 2023, at 21:28, Dong Lin <lindon...@gmail.com> wrote: > > > > >> > > > > >> Hi Piotr, > > > > >> > > > > >> Thanks for the explanation. > > > > >> > > > > >> To recap our offline discussion, there is a concern regarding the > > > > >> capability to dynamically switch between stream and batch modes. > > This > > > > >> concern is around unforeseen behaviors such as bugs or performance > > > > >> regressions, which we might not yet be aware of yet. The reason > for > > > this > > > > >> concern is that this feature involves a fundamental impact on the > > > Flink > > > > >> runtime's behavior. > > > > >> > > > > >> Due to the above concern, I agree it is reasonable to annotate > > related > > > > > APIs > > > > >> as experimental. This step would provide us with the flexibility > to > > > > modify > > > > >> these APIs if issues arise in the future. This annotation also > > serves > > > > as a > > > > >> note to users that this functionality might not perform well as > > > > expected. > > > > >> > > > > >> Though I believe that we can ensure the reliability of this > feature > > > > > through > > > > >> good design and code reviews, comprehensive unit tests, and > thorough > > > > >> integration testing, I agree that it is reasonable to be extra > > > cautious > > > > in > > > > >> this case. Also, it should be OK to delay making these APIs as > > > > >> non-experimental by 1-2 releases. > > > > >> > > > > >> I have updated FLIP-327, FLIP-328, and FLIP-331 to mark APIs in > > these > > > > docs > > > > >> as experimental. Please let me know if you think any other API > > should > > > > also > > > > >> be marked as experimental. > > > > >> > > > > >> Thanks! > > > > >> Dong > > > > >> > > > > >> On Wed, Aug 16, 2023 at 10:39 PM Piotr Nowojski < > > > > piotr.nowoj...@gmail.com> > > > > >> wrote: > > > > >> > > > > >>> Hi Dong, > > > > >>> > > > > >>> Operators API is unfortunately also our public facing API and I > > mean > > > > the > > > > >>> APIs that we will add there should also be marked `@Experimental` > > > IMO. > > > > >>> > > > > >>> The config options should also be marked as experimental (both > > > > >>> annotated @Experimental and noted the same thing in the docs, > > > > >>> if @Experimental annotation is not automatically mentioned in the > > > > docs). > > > > >>> > > > > >>>> Alternatively, how about we add a doc for > > > > >>> checkpointing.interval-during-backlog explaining its > impact/concern > > > as > > > > >>> discussed above? > > > > >>> > > > > >>> We should do this independently from marking the APIs/config > > options > > > as > > > > >>> `@Experimental` > > > > >>> > > > > >>> Best, > > > > >>> Piotrek > > > > >>> > > > > >>> pt., 11 sie 2023 o 14:55 Dong Lin <lindon...@gmail.com> > > napisał(a): > > > > >>> > > > > >>>> Hi Piotr, > > > > >>>> > > > > >>>> Thanks for the reply! > > > > >>>> > > > > >>>> On Fri, Aug 11, 2023 at 4:44 PM Piotr Nowojski < > > > > piotr.nowoj...@gmail.com > > > > >>>> > > > > >>>> wrote: > > > > >>>> > > > > >>>>> Hi, > > > > >>>>> > > > > >>>>> Sorry for the long delay in responding! > > > > >>>>> > > > > >>>>>> Given that it is an optional feature that can be > > > > >>>>>> turned off by users, it might be OK to just let users try it > out > > > and > > > > >>> we > > > > >>>>> can > > > > >>>>>> fix performance issues once we detect any of them. What do you > > > > think? > > > > >>>>> I think it's fine. It would be best to mark this feature as > > > > >>> experimental, > > > > >>>>> and > > > > >>>>> we say that the config keys or the default values might change > in > > > the > > > > >>>>> future. > > > > >>>>> > > > > >>>> In general I agree we can mark APIs that determine "whether to > > > enable > > > > >>>> dynamic switching between stream/batch mode" as experimental. > > > > >>>> > > > > >>>> However, I am not sure we have such an API yet. The APIs added > in > > > this > > > > >>> FLIP > > > > >>>> are intended to be used by operator developers rather than end > > > users. > > > > > End > > > > >>>> users can enable this capability by setting > > > > >>>> execution.checkpointing.interval-during-backlog = Long.MAX and > > uses > > > a > > > > >>>> source which might implicitly set backlog statu (e.g. > > HybridSource). > > > > So > > > > >>>> execution.checkpointing.interval-during-backlog is the only > > > > user-facing > > > > >>>> APIs that can always control whether this feature can be used. > > > > >>>> > > > > >>>> However, execution.checkpointing.interval-during-backlog itself > is > > > not > > > > >>> tied > > > > >>>> to FLIP-327. > > > > >>>> > > > > >>>> Do you mean we should set checkpointing.interval-during-backlog > as > > > > >>>> experimental? Alternatively, how about we add a doc for > > > > >>>> checkpointing.interval-during-backlog explaining its > > impact/concern > > > as > > > > >>>> discussed above? > > > > >>>> > > > > >>>> Best, > > > > >>>> Dong > > > > >>>> > > > > >>>> > > > > >>>>>> Maybe we can revisit the need for such a config when we > > > > >>>> introduce/discuss > > > > >>>>>> the capability to switch backlog from false to true in the > > future. > > > > >>> What > > > > >>>>> do > > > > >>>>>> you think? > > > > >>>>> Sure, we can do that. > > > > >>>>> > > > > >>>>> Best, > > > > >>>>> Piotrek > > > > >>>>> > > > > >>>>> niedz., 23 lip 2023 o 14:32 Dong Lin <lindon...@gmail.com> > > > > napisał(a): > > > > >>>>> > > > > >>>>>> Hi Piotr, > > > > >>>>>> > > > > >>>>>> Thanks a lot for the explanation. Please see my reply inline. > > > > >>>>>> > > > > >>>>>> On Fri, Jul 21, 2023 at 10:49 PM Piotr Nowojski < > > > > >>>>> piotr.nowoj...@gmail.com> > > > > >>>>>> wrote: > > > > >>>>>> > > > > >>>>>>> Hi Dong, > > > > >>>>>>> > > > > >>>>>>> Thanks a lot for the answers. I can now only briefly answer > > your > > > > >>> last > > > > >>>>>>> email. > > > > >>>>>>> > > > > >>>>>>>> It is possible that spilling to disks might cause larger > > > > >>> overhead. > > > > >>>>> IMO > > > > >>>>>> it > > > > >>>>>>>> is an orthogonal issue already existing in Flink. This is > > > > >>> because a > > > > >>>>>> Flink > > > > >>>>>>>> job running batch mode might also be slower than its > > throughput > > > > >>> in > > > > >>>>>> stream > > > > >>>>>>>> mode due to the same reason. > > > > >>>>>>> Yes, I know, but the thing that worries me is that previously > > > only > > > > >>> a > > > > >>>>> user > > > > >>>>>>> alone > > > > >>>>>>> could decide whether to use batch mode or streaming, and in > > > > >>> practice > > > > >>>>> one > > > > >>>>>>> user would rarely (if ever) use both for the same > > > > >>> problem/job/query. > > > > >>>> If > > > > >>>>>> his > > > > >>>>>>> intention was to eventually process live data, he was using > > > > >>> streaming > > > > >>>>>> even > > > > >>>>>>> if there was a large backlog at the start (apart of some very > > few > > > > >>>> very > > > > >>>>>>> power > > > > >>>>>>> users). > > > > >>>>>>> > > > > >>>>>>> With this change, we want to introduce a mode that would be > > > > >>> switching > > > > >>>>>> back > > > > >>>>>>> and forth between streaming and "batch in streaming" > > > automatically. > > > > >>>> So > > > > >>>>> a > > > > >>>>>>> potential performance regression would be much more visible > and > > > > >>>> painful > > > > >>>>>>> at the same time. If batch query runs slower then it could, > > it's > > > > >>> kind > > > > >>>>> of > > > > >>>>>>> fine as > > > > >>>>>>> it will end at some point. If streaming query during large > back > > > > >>>>> pressure > > > > >>>>>>> maybe > > > > >>>>>>> temporary load spike switches to batch processing, that's a > > > bigger > > > > >>>>> deal. > > > > >>>>>>> Especially if batch processing mode will not be able to > > actually > > > > >>> even > > > > >>>>>>> handle > > > > >>>>>>> the normal load, after the load spike. In that case, the job > > > could > > > > >>>>> never > > > > >>>>>>> recover > > > > >>>>>>> from the backpressure/backlog mode. > > > > >>>>>>> > > > > >>>>>> I understand you are concerned with the risk of performance > > > > >>> regression > > > > >>>>>> introduced due to switching to batch mode. > > > > >>>>>> > > > > >>>>>> After thinking about this more, I think this existing proposal > > > meets > > > > >>>> the > > > > >>>>>> minimum requirement of "not introducing regression for > existing > > > > >>> jobs". > > > > >>>>> The > > > > >>>>>> reason is that even if batch mode can be slower than stream > mode > > > for > > > > >>>> some > > > > >>>>>> operators in some cases, this is an optional feature that will > > > only > > > > >>> be > > > > >>>>>> enabled if a user explicitly overrides the newly introduced > > config > > > > to > > > > >>>>>> non-default values. Existing jobs that simply upgrade their > > Flink > > > > >>>> library > > > > >>>>>> version will not suffer any performance regression. > > > > >>>>>> > > > > >>>>>> More specifically, in order to switch to batch mode, users > will > > > need > > > > >>> to > > > > >>>>>> explicitly set execution.checkpointing.interval-during-backlog > > to > > > 0. > > > > >>>> And > > > > >>>>>> users can always explicitly update > > > > >>>>>> execution.checkpointing.interval-during-backlog to turn off > the > > > > batch > > > > >>>>> mode > > > > >>>>>> if that incurs any performance issue. > > > > >>>>>> > > > > >>>>>> As far as I can tell, for all practical workloads we see in > > > > >>> production > > > > >>>>>> jobs, batch mode is always faster (w.r.t. throughput) than > > stream > > > > >>> mode > > > > >>>>> when > > > > >>>>>> there is a high backlog of incoming records. Though it is > still > > > > >>>>>> theoretically possible, it should be very rare (if any) for > > batch > > > > >>> mode > > > > >>>> to > > > > >>>>>> be slower in practice. Given that it is an optional feature > that > > > can > > > > >>> be > > > > >>>>>> turned off by users, it might be OK to just let users try it > out > > > and > > > > >>> we > > > > >>>>> can > > > > >>>>>> fix performance issues once we detect any of them. What do you > > > > think? > > > > >>>>>> > > > > >>>>>> > > > > >>>>>>>> execution.backlog.use-full-batch-mode-on-start (default > false) > > > > >>>>>>> ops sorry, it was supposed to be sth like: > > > > >>>>>>> > > > > >>>>>>> execution.backlog.use-batch-mode-only-on-start (default > false) > > > > >>>>>>> > > > > >>>>>>> That option would disallow switching from streaming to batch. > > > Batch > > > > >>>>> mode > > > > >>>>>>> would be allowed only to get rid of the initial, present on > > > > >>> start-up > > > > >>>>>>> backlog. > > > > >>>>>>> > > > > >>>>>>> Would allow us to safely experiment with switching from > > streaming > > > > >>> to > > > > >>>>>> batch > > > > >>>>>>> and I would be actually more fine in enabling "using batch > mode > > > on > > > > >>>>> start" > > > > >>>>>>> by default, until we gain confidence and feedback that > > switching > > > > >>>> back & > > > > >>>>>>> forth > > > > >>>>>>> is working as expected. > > > > >>>>>>> > > > > >>>>>> Now I understand what you are suggesting. I agree that it is > > > > >>> necessary > > > > >>>>> for > > > > >>>>>> users to be able to disallow switching from streaming to > batch. > > > > >>>>>> > > > > >>>>>> I am not sure it is necessary to introduce an extra config > just > > > for > > > > >>>> this > > > > >>>>>> purpose. The reason is that we don't have any strategy that > > > switches > > > > >>>>>> backlog status from false to true yet. And when we have such > > > > strategy > > > > >>>>> (e.g. > > > > >>>>>> FLIP-328) in the future, it is very likely that we will > > introduce > > > > >>> extra > > > > >>>>>> config(s) for users to explicitly turn on such a feature. That > > > means > > > > >>>> user > > > > >>>>>> should be able to turn off this feature even if we don't have > > > > >>> something > > > > >>>>>> like execution.backlog.use-batch-mode-only-on-start. > > > > >>>>>> > > > > >>>>>> Maybe we can revisit the need for such a config when we > > > > >>>> introduce/discuss > > > > >>>>>> the capability to switch backlog from false to true in the > > future. > > > > >>> What > > > > >>>>> do > > > > >>>>>> you think? > > > > >>>>>> > > > > >>>>>> > > > > >>>>>>>>> Or we could limit the scope of this FLIP to only support > > > > >>> starting > > > > >>>>> with > > > > >>>>>>>>> batch mode and switching only once to > > > > >>>>>>>>> streaming, and design a follow up with switching back and > > > forth? > > > > >>>>>>>> Sure, that sounds good to me. I am happy to split this FLIP > > into > > > > >>>> two > > > > >>>>>>> FLIPs > > > > >>>>>>>> so that we can make incremental progress. > > > > >>>>>>> Great, let's do that. In a follow up FLIP we could restart > the > > > > >>>>> discussion > > > > >>>>>>> about > > > > >>>>>>> switching back and forth. > > > > >>>>>>> > > > > >>>>>> Cool, I added the following statement to the motivation > section. > > > > >>>>>> > > > > >>>>>> "NOTE: this FLIP focuses only on the capability to switch from > > > batch > > > > >>> to > > > > >>>>>> stream mode. If there is any extra API needed to support > > switching > > > > >>> from > > > > >>>>>> stream to batch mode, we will discuss them in a follow-up > FLIP." > > > > >>>>>> > > > > >>>>>> I am looking forward to reading your follow-up thoughts! > > > > >>>>>> > > > > >>>>>> Best, > > > > >>>>>> Dong > > > > >>>>>> > > > > >>>>>> > > > > >>>>>>> Piotrek > > > > >>>>>>> > > > > >>>>>>> czw., 20 lip 2023 o 16:57 Dong Lin <lindon...@gmail.com> > > > > >>> napisał(a): > > > > >>>>>>>> Hi Piotr, > > > > >>>>>>>> > > > > >>>>>>>> Thank you for the very detailed comments! Please see my > reply > > > > >>>> inline. > > > > >>>>>>>> On Thu, Jul 20, 2023 at 12:24 AM Piotr Nowojski < > > > > >>>>>>> piotr.nowoj...@gmail.com> > > > > >>>>>>>> wrote: > > > > >>>>>>>> > > > > >>>>>>>>> Hi Dong, > > > > >>>>>>>>> > > > > >>>>>>>>> I have a couple of follow up questions about switching back > > and > > > > >>>>> forth > > > > >>>>>>>>> between streaming and batching mode. > > > > >>>>>>>>> Especially around shuffle/watermark strategy, and keyed > state > > > > >>>>>> backend. > > > > >>>>>>>>> First of all, it might not always be beneficial to switch > > into > > > > >>>> the > > > > >>>>>>> batch > > > > >>>>>>>>> modes: > > > > >>>>>>>>> - Shuffle strategy > > > > >>>>>>>>> - Is sorting going to be purely in-memory? If not, > > > > >>> obviously > > > > >>>>>>> spilling > > > > >>>>>>>>> to disks might cause larger overheads > > > > >>>>>>>>> compared to not sorting the records. > > > > >>>>>>>>> > > > > >>>>>>>> Sorting might require spilling data to disk depending on the > > > > >>> input > > > > >>>>>> size. > > > > >>>>>>>> The behavior of sorting w.r.t. memory/disk is expected to be > > > > >>>> exactly > > > > >>>>>> the > > > > >>>>>>>> same as the behavior of input sorting automatically > performed > > by > > > > >>>>> Flink > > > > >>>>>>>> runtime in batch mode for keyed inputs. > > > > >>>>>>>> > > > > >>>>>>>> More specifically, ExternalSorter > > > > >>>>>>>> < > > > > >>>>>>>> > > > > > > > > > > > > > > > https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/ExternalSorter.java > > > > >>>>>>>> is > > > > >>>>>>>> currently used to sort keyed inputs in batch mode. It is > > > > >>>>> automatically > > > > >>>>>>> used > > > > >>>>>>>> by Flink runtime in OneInputStreamTask (here > > > > >>>>>>>> < > > > > >>>>>>>> > > > > > > > > > > > > > > > https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTask.java#L114 > > > > >>>>>>>>> ) > > > > >>>>>>>> and in MultiInputSortingDataInput (here > > > > >>>>>>>> < > > > > >>>>>>>> > > > > > > > > > > > > > > > https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/MultiInputSortingDataInput.java#L188 > > > > >>>>>>>>> ). > > > > >>>>>>>> We plan to re-use the same code/mechanism to do sorting. > > > > >>>>>>>> > > > > >>>>>>>> It is possible that spilling to disks might cause larger > > > > >>> overhead. > > > > >>>>> IMO > > > > >>>>>> it > > > > >>>>>>>> is an orthogonal issue already existing in Flink. This is > > > > >>> because a > > > > >>>>>> Flink > > > > >>>>>>>> job running batch mode might also be slower than its > > throughput > > > > >>> in > > > > >>>>>> stream > > > > >>>>>>>> mode due to the same reason. However, even though it is > > possible > > > > >>> in > > > > >>>>>>> theory, > > > > >>>>>>>> I expect that in practice the throughput of using sorting + > > > > >>>>>>>> BatchExecutionKeyedStateBackend should be much higher than > > using > > > > >>>>> other > > > > >>>>>>>> keyed statebackends when the amount of data is large. As a > > > matter > > > > >>>> of > > > > >>>>>>> fact, > > > > >>>>>>>> we have not heard of complaints of such performance > regression > > > > >>>> issues > > > > >>>>>> in > > > > >>>>>>>> batch mode. > > > > >>>>>>>> > > > > >>>>>>>> The primary goal of this FLIP is to allow the operator to > run > > at > > > > >>>> the > > > > >>>>>> same > > > > >>>>>>>> throughput (in stream mode when there is backlog) as it can > > > > >>>> currently > > > > >>>>>> do > > > > >>>>>>> in > > > > >>>>>>>> batch mode. And this goal is not affected by the disk > overhead > > > > >>>> issue > > > > >>>>>>>> mentioned above. > > > > >>>>>>>> > > > > >>>>>>>> I am thinking maybe we can treat it as an orthogonal > > performance > > > > >>>>>>>> optimization problem instead of solving this problem in this > > > > >>> FLIP? > > > > >>>>>>>> - If it will be at least partially in-memory, does Flink > > > have > > > > >>>>> some > > > > >>>>>>>>> mechanism to reserve optional memory that > > > > >>>>>>>>> can be revoked if a new operator starts up? Can this > > > > >>> memory > > > > >>>>> be > > > > >>>>>>>>> redistributed? Ideally we should use as > > > > >>>>>>>>> much as possible of the available memory to avoid > > > > >>> spilling > > > > >>>>>> costs, > > > > >>>>>>>> but > > > > >>>>>>>>> also being able to revoke that memory > > > > >>>>>>>>> > > > > >>>>>>>> This FLIP does not support dynamically > revoking/redistribuitng > > > > >>>>> managed > > > > >>>>>>>> memory used by the ExternalSorter. > > > > >>>>>>>> > > > > >>>>>>>> For operators with isInternalSorterSupported = true, we will > > > > >>>> allocate > > > > >>>>>> to > > > > >>>>>>>> this operator execution.sorted-inputs.memory > > > > >>>>>>>> < > > > > >>>>>>>> > > > > > > > > > > > > > > > https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/configuration/ExecutionOptions.java#L144 > > > > >>>>>>>> amount of managed memory. This is the same as how Flink > > > allocates > > > > >>>>>> managed > > > > >>>>>>>> memory to an operator when this operator has keyed inputs in > > > > >>> batch > > > > >>>>>> mode. > > > > >>>>>>>> Note that this FLIP intends to support operators to sort > > inputs > > > > >>>>>> whenever > > > > >>>>>>>> there is backlog. And there is currently no way for an > > operator > > > > >>> to > > > > >>>>> know > > > > >>>>>>> in > > > > >>>>>>>> advance whether there will be no backlog after a given time. > > So > > > > >>> it > > > > >>>>>> seems > > > > >>>>>>>> simpler to just keep managed memory for such an operator > > > > >>> throughout > > > > >>>>> the > > > > >>>>>>>> lifecycle of this operator, for now. > > > > >>>>>>>> > > > > >>>>>>>> Besides, it seems that the lack of ability to dynamically > > > > >>>>>>>> revoke/redistribute un-used managed memory is an existing > > issue > > > > >>> in > > > > >>>>>> Flink. > > > > >>>>>>>> For example, we might have two operators sharing the same > slot > > > > >>> and > > > > >>>>>> these > > > > >>>>>>>> two operators both use managed memory (e.g. to sort inputs). > > > > >>> There > > > > >>>> is > > > > >>>>>>>> currently no way for one operator to re-use the memory not > > used > > > > >>> by > > > > >>>>> the > > > > >>>>>>>> other operator. > > > > >>>>>>>> > > > > >>>>>>>> Therefore, I think we can treat this as an orthogonal > > > performance > > > > >>>>>>>> optimization problem which can be addressed separately. What > > do > > > > >>> you > > > > >>>>>>> think? > > > > >>>>>>>> > > > > >>>>>>>>> - Sometimes sorting, even if we have memory to do that, > > > > >>> might > > > > >>>>> be > > > > >>>>>> an > > > > >>>>>>>>> unnecessary overhead. > > > > >>>>>>>>> - Watermarks > > > > >>>>>>>>> - Is holding back watermarks always good? If we have > tons > > > > >>> of > > > > >>>>> data > > > > >>>>>>>>> buffered/sorted and waiting to be processed > > > > >>>>>>>>> with multiple windows per key and many different > keys. > > > > >>>> When > > > > >>>>> we > > > > >>>>>>>>> switch back to `isBacklog=false` we > > > > >>>>>>>>> first process all of that data before processing > > > > >>>> watermarks, > > > > >>>>>> for > > > > >>>>>>>>> operators that are not using sorted input the > > > > >>>>>>>>> state size can explode significantly causing lots of > > > > >>>>> problems. > > > > >>>>>>>> Even > > > > >>>>>>>>> for those that can use sorting, switching to > > > > >>>>>>>>> sorting or BatchExecutionKeyedStateBackend is not > > > > >>> always a > > > > >>>>>> good > > > > >>>>>>>>> idea, but keeping RocksDB also can be > > > > >>>>>>>>> risky. > > > > >>>>>>>>> > > > > >>>>>>>> With the current FLIP, the proposal is to use a sorter only > > when > > > > >>>> the > > > > >>>>>>> inputs > > > > >>>>>>>> have keys. According to this practice, operators which are > not > > > > >>>> using > > > > >>>>>>>> sorting should have un-keyed inputs. I believe such an > > operator > > > > >>>> will > > > > >>>>>> not > > > > >>>>>>>> even use a keyed state backend. Maybe I missed some > use-case. > > > Can > > > > >>>> you > > > > >>>>>>>> provide a use-case where we will have an operator with > > un-keyed > > > > >>>>> inputs > > > > >>>>>>>> whose state size can explode due to we holding back > > watermarks? > > > > >>>>>>>> > > > > >>>>>>>> For operators with keyed inputs that use sorting, I suppose > it > > > is > > > > >>>>>>> possible > > > > >>>>>>>> that sorting + BatchExecutionKeyedStateBackend can be worse > > than > > > > >>>>> using > > > > >>>>>>>> RocksDB. But I believe this is very very rare (if possible) > in > > > > >>>> almost > > > > >>>>>>>> practical usage of Flink. > > > > >>>>>>>> > > > > >>>>>>>> Take one step back, if this indeed cause regression for a > real > > > > >>>>>> use-case, > > > > >>>>>>>> user can set execution.checkpointing.interval-during-backlog > > to > > > > >>>>>> anything > > > > >>>>>>>> other than 0 so that this FLIP will not use > > > > >>>>>>>> sorter + BatchExecutionKeyedStateBackend even even when > there > > is > > > > >>>>>> backlog. > > > > >>>>>>>> I would hope we can find a way to automatically determine > > > whether > > > > >>>>> using > > > > >>>>>>>> sorting + BatchExecutionKeyedStateBackend can be better or > > worse > > > > >>>> than > > > > >>>>>>> using > > > > >>>>>>>> RocksDB alone. But I could not find a good and reliable way > to > > > do > > > > >>>>> this. > > > > >>>>>>>> Maybe we can update Flink to do this when we find a good way > > to > > > > >>> do > > > > >>>>> this > > > > >>>>>>> in > > > > >>>>>>>> the future? > > > > >>>>>>>> > > > > >>>>>>>> > > > > >>>>>>>> > > > > >>>>>>>>> - Keyed state backend > > > > >>>>>>>>> - I think you haven't described what happens during > > > > >>> switching > > > > >>>>>> from > > > > >>>>>>>>> streaming to backlog processing. > > > > >>>>>>>>> > > > > >>>>>>>> Good point. This indeed needs to be described. I added a > TODO > > in > > > > >>>> the > > > > >>>>>>>> "Behavior changes ..." section to describe what happens when > > > > >>>>> isBacklog > > > > >>>>>>>> switches from false to true, for all > > > > >>>>> watermark/checkpoint/statebackend > > > > >>>>>>> etc. > > > > >>>>>>>> Let me explain this for the state backend here for now. I > will > > > > >>>> update > > > > >>>>>>> FLIP > > > > >>>>>>>> later. > > > > >>>>>>>> > > > > >>>>>>>> When isBacklog switches from false to true, operator with > > keyed > > > > >>>>> inputs > > > > >>>>>>> can > > > > >>>>>>>> optionally (as determined by its implementation) starts to > use > > > > >>>>> internal > > > > >>>>>>>> sorter to sort inputs by key, without processing inputs or > > > > >>> updating > > > > >>>>>>>> statebackend, until it receives end-of-inputs or isBacklog > is > > > > >>>>> switched > > > > >>>>>> to > > > > >>>>>>>> false again. > > > > >>>>>>>> > > > > >>>>>>>> > > > > >>>>>>>> > > > > >>>>>>>>> - Switch can be an unnecessary overhead. > > > > >>>>>>>> > > > > >>>>>>>> I agree it can cause unnecessary overhead, particularly when > > > > >>>>> isBacklog > > > > >>>>>>>> switches back and forth frequently. Whether or not this is > > > > >>>>> unnecessary > > > > >>>>>>>> likely depends on the duration/throughput of the backlog > phase > > > as > > > > >>>>> well > > > > >>>>>> as > > > > >>>>>>>> the specific computation logic of the operator. I am not > sure > > > > >>> there > > > > >>>>> is > > > > >>>>>> a > > > > >>>>>>>> good way for Flink to determine in advance whether switching > > is > > > > >>>>>>>> unnecessary. > > > > >>>>>>>> > > > > >>>>>>>> Note that for the existing use-case where we expect to > change > > > > >>>>> isBacklog > > > > >>>>>>> to > > > > >>>>>>>> true (e.g. MySQL CDC snapshot phase, Kafka source watermark > > lag > > > > >>>> being > > > > >>>>>> too > > > > >>>>>>>> high), we don't expect the watermark to switch back and > force > > > > >>>>>> frequently. > > > > >>>>>>>> And user can disable this switch by setting > > > > >>>>>>>> execution.checkpointing.interval-during-backlog to anything > > > other > > > > >>>>> than > > > > >>>>>> 0. > > > > >>>>>>>> Therefore, I am wondering if we can also view this as a > > > > >>> performance > > > > >>>>>>>> optimization opportunity for extra use-cases in the future, > > > > >>> rather > > > > >>>>>> than a > > > > >>>>>>>> blocking issue of this FLIP for the MVP use-case (e.g. > > snapshot > > > > >>>> phase > > > > >>>>>> for > > > > >>>>>>>> any CDC source, Kafka watermark lag). > > > > >>>>>>>> > > > > >>>>>>>> > > > > >>>>>>>>> At the same time, in your current proposal, for > > > > >>>>>>>>> `execution.checkpointing.interval-during-backlog > 0` we > > won't > > > > >>>>>>>>> switch to "batch" mode at all. That's a bit of shame, I > don't > > > > >>>>>>> understand > > > > >>>>>>>>> why those two things should be coupled > > > > >>>>>>>>> together? > > > > >>>>>>>>> > > > > >>>>>>>> We can in general classify optimizations as those that are > > > > >>>> compatible > > > > >>>>>>> with > > > > >>>>>>>> checkpointing, and those that are not compatible with > > > > >>>> checkpointing. > > > > >>>>>> For > > > > >>>>>>>> example, input sorting is currently not compatible with > > > > >>>>> checkpointing. > > > > >>>>>>> And > > > > >>>>>>>> buffering input records to reduce state backend overhead > (and > > > > >>>>> probably > > > > >>>>>>>> columnar processing for mini-batch in the future) is > > compatible > > > > >>>> with > > > > >>>>>>>> checkpointing. > > > > >>>>>>>> > > > > >>>>>>>> The primary of FLIP-327 is to support optimizations not > > > > >>> compatible > > > > >>>>> with > > > > >>>>>>>> checkpointing. If > > > > >>> execution.checkpointing.interval-during-backlog > > > > > >>>>> 0, > > > > >>>>>>>> which means that user intends to still do checkpointing even > > > when > > > > >>>>> there > > > > >>>>>>> is > > > > >>>>>>>> backog, then we will not be able to support such > > optimizations. > > > > >>>>>>>> > > > > >>>>>>>> For optimizations that are compatible with checkpointing, we > > can > > > > >>> do > > > > >>>>>> this > > > > >>>>>>>> even when the operator does not run in "batch mode". There > are > > > > >>>> extra > > > > >>>>>>>> problems to solve in order to achieve this optimization, > such > > as > > > > >>>>>>> supporting > > > > >>>>>>>> unaligned checkpointing without prolonging its sync phase. I > > > plan > > > > >>>> to > > > > >>>>>>>> explain how this can be done in FLIP-325. > > > > >>>>>>>> > > > > >>>>>>>> > > > > >>>>>>>>> All in all, shouldn't we aim for some more clever process > of > > > > >>>>>> switching > > > > >>>>>>>> back > > > > >>>>>>>>> and forth between streaming/batch modes > > > > >>>>>>>>> for watermark strategy/state backend/sorting based on some > > > > >>>> metrics? > > > > >>>>>>>> Trying > > > > >>>>>>>>> to either predict if switching might help, > > > > >>>>>>>>> or trying to estimate if the last switch was beneficial? > > Maybe > > > > >>>>>>> something > > > > >>>>>>>>> along the lines: > > > > >>>>>>>>> - sort only in memory and during sorting count the number > of > > > > >>>>> distinct > > > > >>>>>>>> keys > > > > >>>>>>>>> (NDK) > > > > >>>>>>>>> - maybe allow for spilling if so far in memory we have > > NDK > > > > >>> * > > > > >>>> 5 > > > > >>>>>> = > > > > >>>>>>>>> #records > > > > >>>>>>>>> - do not allow to buffer records above a certain threshold, > > as > > > > >>>>>>> otherwise > > > > >>>>>>>>> checkpointing can explode > > > > >>>>>>>>> - switch to `BatchExecutionKeyedStateBackend` only if NDK > * 2 > > > > >>>> = > > > > >>>>>>> #records > > > > >>>>>>>>> - do not sort if last NDKs (or EMA of NDK?) 1.5 <= #records > > > > >>>>>>>>> > > > > >>>>>>>>> Or even maybe for starters something even simpler and then > > test > > > > >>>> out > > > > >>>>>>>>> something more fancy as a follow up? > > > > >>>>>>>>> > > > > >>>>>>>> I agree it is worth investigating these ideas to further > > > optimize > > > > >>>> the > > > > >>>>>>>> performance during backlog. > > > > >>>>>>>> > > > > >>>>>>>> I just think these can be done independently after this > FLIP. > > > The > > > > >>>>> focus > > > > >>>>>>> of > > > > >>>>>>>> this FLIP is to re-use in stream mode the same optimization > > > which > > > > >>>> we > > > > >>>>>>>> already use in batch mode, rather than inventing or > improving > > > the > > > > >>>>>>>> performance of these existing optimizations. > > > > >>>>>>>> > > > > >>>>>>>> Given that there are already a lot of new mechanism/features > > to > > > > >>>>> discuss > > > > >>>>>>> and > > > > >>>>>>>> address in this FLIP, I am hoping we can limit the scope of > > this > > > > >>>> FLIP > > > > >>>>>> to > > > > >>>>>>>> re-use the existing optimization, and do these extra > > > optimization > > > > >>>>>>>> opportunities as future work. > > > > >>>>>>>> > > > > >>>>>>>> What do you think? > > > > >>>>>>>> > > > > >>>>>>>> > > > > >>>>>>>>> At the same time, > > > > >>>>> `execution.checkpointing.interval-during-backlog=0` > > > > >>>>>>>> seems > > > > >>>>>>>>> a weird setting to me, that I would > > > > >>>>>>>>> not feel safe recommending to anyone. If processing of a > > > > >>> backlog > > > > >>>>>> takes > > > > >>>>>>> a > > > > >>>>>>>>> long time, a job might stop making > > > > >>>>>>>>> any progress due to some random failures. Especially > > dangerous > > > > >>>> if a > > > > >>>>>> job > > > > >>>>>>>> switches from streaming mode back to > > > > >>>>>>>>> backlog processing due to some reasons, as that could > happen > > > > >>>> months > > > > >>>>>>> after > > > > >>>>>>>>> someone started a job with this > > > > >>>>>>>>> strange setting. So should we even have it? I would simply > > > > >>>> disallow > > > > >>>>>>> it. I > > > > >>>>>>>> Good point. I do agree we need to further work to improve > the > > > > >>>>> failover > > > > >>>>>>>> performance in case any task fails. > > > > >>>>>>>> > > > > >>>>>>>> As of the current FLIP, if any task fails during backlog and > > > > >>>>>>>> execution.checkpointing.interval-during-backlog = 0, we will > > > need > > > > >>>> to > > > > >>>>>>>> restart all operators to the last checkpointed state and > > > continue > > > > >>>>>>>> processing backlog. And this can be a lot of rollback since > > > there > > > > >>>> is > > > > >>>>> no > > > > >>>>>>>> checkpoint during backlog. And this can also be worse than > > batch > > > > >>>>> since > > > > >>>>>>> this > > > > >>>>>>>> FLIP currently does not support exporting/saving records to > > > local > > > > >>>>> disk > > > > >>>>>>> (or > > > > >>>>>>>> shuffle service) so that a failed task can re-consume the > > > records > > > > >>>>> from > > > > >>>>>>> the > > > > >>>>>>>> upstream task (or shuffle service) in the same way as how > > Flink > > > > >>>>>> failover > > > > >>>>>>> a > > > > >>>>>>>> task in batch mode. > > > > >>>>>>>> > > > > >>>>>>>> I think we can extend this FLIP to solve this problem so > that > > it > > > > >>>> can > > > > >>>>>> have > > > > >>>>>>>> at least the same behavior/performance as batch-mode job. > The > > > > >>> idea > > > > >>>> is > > > > >>>>>> to > > > > >>>>>>>> also follow what batch mode does. For example, we can > trigger > > a > > > > >>>>>>> checkpoint > > > > >>>>>>>> when isBacklog switches to true, and every operator should > > > buffer > > > > >>>> its > > > > >>>>>>>> output in the TM local disk (or remote shuffle service). > > > > >>> Therefore, > > > > >>>>>>> after a > > > > >>>>>>>> task fails, it can restart from the last checkpoint and > > > > >>> re-consume > > > > >>>>> data > > > > >>>>>>>> buffered in the upstream task. > > > > >>>>>>>> > > > > >>>>>>>> I will update FLIP as described above. Would this address > your > > > > >>>>> concern? > > > > >>>>>>>> > > > > >>>>>>>> > > > > >>>>>>>>> could see a power setting like: > > > > >>>>>>>>> `execution.backlog.use-full-batch-mode-on-start > > > > >>> (default > > > > >>>>>>> false)` > > > > >>>>>>>> I am not sure I fully understand this config or its > > motivation. > > > > >>> Can > > > > >>>>> you > > > > >>>>>>>> help explain the exact semantics of this config? > > > > >>>>>>>> > > > > >>>>>>>> > > > > >>>>>>>>> that would override any heuristic of switching to backlog > if > > > > >>>>> someone > > > > >>>>>> is > > > > >>>>>>>>> submitting a new job that starts with > > > > >>>>>>>>> `isBacklog=true`. > > > > >>>>>>>>> > > > > >>>>>>>>> Or we could limit the scope of this FLIP to only support > > > > >>> starting > > > > >>>>>> with > > > > >>>>>>>>> batch mode and switching only once to > > > > >>>>>>>>> streaming, and design a follow up with switching back and > > > > >>> forth? > > > > >>>>>>>> Sure, that sounds good to me. I am happy to split this FLIP > > into > > > > >>>> two > > > > >>>>>>> FLIPs > > > > >>>>>>>> so that we can make incremental progress. > > > > >>>>>>>> > > > > >>>>>>>> Best, > > > > >>>>>>>> Dong > > > > >>>>>>>> > > > > >>>>>>>> > > > > >>>>>>>>> I'm looking forwards to hearing/reading out your thoughts. > > > > >>>>>>>>> > > > > >>>>>>>>> Best, > > > > >>>>>>>>> Piotrek > > > > >>>>>>>>> > > > > >>>>>>>>> > > > > >>>>>>>>> śr., 12 lip 2023 o 12:38 Jing Ge > <j...@ververica.com.invalid > > > > > > > >>>>>>>> napisał(a): > > > > >>>>>>>>>> Hi Dong, > > > > >>>>>>>>>> > > > > >>>>>>>>>> Thanks for your reply! > > > > >>>>>>>>>> > > > > >>>>>>>>>> Best regards, > > > > >>>>>>>>>> Jing > > > > >>>>>>>>>> > > > > >>>>>>>>>> On Wed, Jul 12, 2023 at 3:25 AM Dong Lin < > > > > >>> lindon...@gmail.com> > > > > >>>>>>> wrote: > > > > >>>>>>>>>>> Hi Jing, > > > > >>>>>>>>>>> > > > > >>>>>>>>>>> Thanks for the comments. Please see my reply inline. > > > > >>>>>>>>>>> > > > > >>>>>>>>>>> On Wed, Jul 12, 2023 at 5:04 AM Jing Ge > > > > >>>>>> <j...@ververica.com.invalid > > > > >>>>>>>>>>> wrote: > > > > >>>>>>>>>>> > > > > >>>>>>>>>>>> Hi Dong, > > > > >>>>>>>>>>>> > > > > >>>>>>>>>>>> Thanks for the clarification. Now it is clear for me. I > > > > >>> got > > > > >>>>>>>>> additional > > > > >>>>>>>>>>> noob > > > > >>>>>>>>>>>> questions wrt the internal sorter. > > > > >>>>>>>>>>>> > > > > >>>>>>>>>>>> 1. when to call setter to set the > internalSorterSupported > > > > >>>> to > > > > >>>>> be > > > > >>>>>>>> true? > > > > >>>>>>>>>>> Developer of the operator class (i.e. those classes which > > > > >>>>>>> implements > > > > >>>>>>>>>>> `StreamOperator`) should override the > > > > >>>>> `#getOperatorAttributes()` > > > > >>>>>>> API > > > > >>>>>>>> to > > > > >>>>>>>>>> set > > > > >>>>>>>>>>> internalSorterSupported to true, if he/she decides to > sort > > > > >>>>>> records > > > > >>>>>>>>>>> internally in the operator. > > > > >>>>>>>>>>> > > > > >>>>>>>>>>> > > > > >>>>>>>>>>>> 2 > > > > >>>>>>>>>>>> *"For those operators whose throughput can be > > > > >>> considerably > > > > >>>>>>> improved > > > > >>>>>>>>>> with > > > > >>>>>>>>>>> an > > > > >>>>>>>>>>>> internal sorter, update it to take advantage of the > > > > >>>> internal > > > > >>>>>>> sorter > > > > >>>>>>>>>> when > > > > >>>>>>>>>>>> its input has isBacklog=true.* > > > > >>>>>>>>>>>> *Typically, operators that involve aggregation operation > > > > >>>>> (e.g. > > > > >>>>>>>> join, > > > > >>>>>>>>>>>> cogroup, aggregate) on keyed inputs can benefit from > > > > >>> using > > > > >>>> an > > > > >>>>>>>>> internal > > > > >>>>>>>>>>>> sorter."* > > > > >>>>>>>>>>>> > > > > >>>>>>>>>>>> *"The operator that performs CoGroup operation will > > > > >>>>> instantiate > > > > >>>>>>> two > > > > >>>>>>>>>>>> internal sorter to sorts records from its two inputs > > > > >>>>>> separately. > > > > >>>>>>>> Then > > > > >>>>>>>>>> it > > > > >>>>>>>>>>>> can pull the sorted records from these two sorters. This > > > > >>>> can > > > > >>>>> be > > > > >>>>>>>> done > > > > >>>>>>>>>>>> without wrapping input records with TaggedUnion<...>. In > > > > >>>>>>>> comparison, > > > > >>>>>>>>>> the > > > > >>>>>>>>>>>> existing DataStream#coGroup needs to wrap input records > > > > >>>> with > > > > >>>>>>>>>>>> TaggedUnion<...> before sorting them using one external > > > > >>>>> sorter, > > > > >>>>>>>> which > > > > >>>>>>>>>>>> introduces higher overhead."* > > > > >>>>>>>>>>>> > > > > >>>>>>>>>>>> According to the performance test, it seems that > internal > > > > >>>>>> sorter > > > > >>>>>>>> has > > > > >>>>>>>>>>> better > > > > >>>>>>>>>>>> performance than external sorter. Is it possible to make > > > > >>>>> those > > > > >>>>>>>>>> operators > > > > >>>>>>>>>>>> that can benefit from it use internal sorter by default? > > > > >>>>>>>>>>>> > > > > >>>>>>>>>>> Yes, it is possible. After this FLIP is done, users can > use > > > > >>>>>>>>>>> DataStream#coGroup with EndOfStreamWindows as the window > > > > >>>>> assigner > > > > >>>>>>> to > > > > >>>>>>>>>>> co-group two streams in effectively the batch manner. An > > > > >>>>> operator > > > > >>>>>>>> that > > > > >>>>>>>>>> uses > > > > >>>>>>>>>>> an internal sorter will be used to perform the co-group > > > > >>>>>> operation. > > > > >>>>>>>>> There > > > > >>>>>>>>>> is > > > > >>>>>>>>>>> no need for users of the DataStream API to explicitly > know > > > > >>> or > > > > >>>>> set > > > > >>>>>>> the > > > > >>>>>>>>>>> internal sorter in anyway. > > > > >>>>>>>>>>> > > > > >>>>>>>>>>> In the future, we plan to incrementally optimize other > > > > >>>>>> aggregation > > > > >>>>>>>>>>> operation (e.g. aggregate) on the DataStream API when > > > > >>>>>>>>> EndOfStreamWindows > > > > >>>>>>>>>> is > > > > >>>>>>>>>>> used as the window assigner. > > > > >>>>>>>>>>> > > > > >>>>>>>>>>> Best, > > > > >>>>>>>>>>> Dong > > > > >>>>>>>>>>> > > > > >>>>>>>>>>> > > > > >>>>>>>>>>>> Best regards, > > > > >>>>>>>>>>>> Jing > > > > >>>>>>>>>>>> > > > > >>>>>>>>>>>> > > > > >>>>>>>>>>>> On Tue, Jul 11, 2023 at 2:58 PM Dong Lin < > > > > >>>>> lindon...@gmail.com> > > > > >>>>>>>>> wrote: > > > > >>>>>>>>>>>>> Hi Jing, > > > > >>>>>>>>>>>>> > > > > >>>>>>>>>>>>> Thank you for the comments! Please see my reply inline. > > > > >>>>>>>>>>>>> > > > > >>>>>>>>>>>>> On Tue, Jul 11, 2023 at 5:41 AM Jing Ge > > > > >>>>>>>> <j...@ververica.com.invalid > > > > >>>>>>>>>>>>> wrote: > > > > >>>>>>>>>>>>> > > > > >>>>>>>>>>>>>> Hi Dong, > > > > >>>>>>>>>>>>>> > > > > >>>>>>>>>>>>>> Thanks for the proposal! The FLIP is already in good > > > > >>>>>> shape. I > > > > >>>>>>>> got > > > > >>>>>>>>>>> some > > > > >>>>>>>>>>>>> NIT > > > > >>>>>>>>>>>>>> questions. > > > > >>>>>>>>>>>>>> > > > > >>>>>>>>>>>>>> 1. It is a little bit weird to write the hint right > > > > >>>> after > > > > >>>>>> the > > > > >>>>>>>>>>>> motivation > > > > >>>>>>>>>>>>>> that some features have been moved to FLIP-331, > > > > >>> because > > > > >>>>> at > > > > >>>>>>> that > > > > >>>>>>>>>> time, > > > > >>>>>>>>>>>>>> readers don't know the context about what features > > > > >>> does > > > > >>>>> it > > > > >>>>>>>> mean. > > > > >>>>>>>>> I > > > > >>>>>>>>>>>> would > > > > >>>>>>>>>>>>>> suggest moving the note to the beginning of "Public > > > > >>>>>>> interfaces" > > > > >>>>>>>>>>>> sections. > > > > >>>>>>>>>>>>> Given that the reviewer who commented on this email > > > > >>>> thread > > > > >>>>>>>> before I > > > > >>>>>>>>>>>>> refactored the FLIP (i.e. Piotr) has read FLP-331, I > > > > >>>> think > > > > >>>>> it > > > > >>>>>>> is > > > > >>>>>>>>>>> simpler > > > > >>>>>>>>>>>> to > > > > >>>>>>>>>>>>> just remove any mention of FLIP-331. I have updated the > > > > >>>>> FLIP > > > > >>>>>>>>>>> accordingly. > > > > >>>>>>>>>>>>> > > > > >>>>>>>>>>>>>> 2. It is also a little bit weird to describe all > > > > >>>>> behaviour > > > > >>>>>>>>> changes > > > > >>>>>>>>>> at > > > > >>>>>>>>>>>>> first > > > > >>>>>>>>>>>>>> but only focus on one single feature, i.e. how to > > > > >>>>> implement > > > > >>>>>>>>>>>>>> internalSorterSupported. TBH, I was lost while I was > > > > >>>>>> reading > > > > >>>>>>>> the > > > > >>>>>>>>>>> Public > > > > >>>>>>>>>>>>>> interfaces. Maybe change the FLIP title? Another > > > > >>> option > > > > >>>>>> could > > > > >>>>>>>> be > > > > >>>>>>>>> to > > > > >>>>>>>>>>>>> write a > > > > >>>>>>>>>>>>>> short summary of all features and point out that this > > > > >>>>> FLIP > > > > >>>>>>> will > > > > >>>>>>>>>> only > > > > >>>>>>>>>>>>> focus > > > > >>>>>>>>>>>>>> on the internalSorterSupported feature. Others could > > > > >>> be > > > > >>>>>> found > > > > >>>>>>>> in > > > > >>>>>>>>>>>>> FLIP-331. > > > > >>>>>>>>>>>>>> WDYT? > > > > >>>>>>>>>>>>>> > > > > >>>>>>>>>>>>> Conceptually, the purpose of this FLIP is to allow a > > > > >>>> stream > > > > >>>>>>> mode > > > > >>>>>>>>> job > > > > >>>>>>>>>> to > > > > >>>>>>>>>>>> run > > > > >>>>>>>>>>>>> parts of the topology in batch mode so that it can > > > > >>> apply > > > > >>>>>>>>>>>>> optimizations/computations that can not be used > > > > >>> together > > > > >>>>> with > > > > >>>>>>>>>>>> checkpointing > > > > >>>>>>>>>>>>> (and thus not usable in stream mode). Although internal > > > > >>>>>> sorter > > > > >>>>>>> is > > > > >>>>>>>>> the > > > > >>>>>>>>>>>> only > > > > >>>>>>>>>>>>> optimization immediately supported in this FLIP, this > > > > >>>> FLIP > > > > >>>>>> lays > > > > >>>>>>>> the > > > > >>>>>>>>>>>>> foundation to support other optimizations in the > > > > >>> future, > > > > >>>>> such > > > > >>>>>>> as > > > > >>>>>>>>>> using > > > > >>>>>>>>>>>> GPU > > > > >>>>>>>>>>>>> to process a bounded stream of records. > > > > >>>>>>>>>>>>> > > > > >>>>>>>>>>>>> Therefore, I find it better to keep the current title > > > > >>>>> rather > > > > >>>>>>> than > > > > >>>>>>>>>>>> limiting > > > > >>>>>>>>>>>>> the scope to internal sorter. What do you think? > > > > >>>>>>>>>>>>> > > > > >>>>>>>>>>>>> > > > > >>>>>>>>>>>>> > > > > >>>>>>>>>>>>>> 3. There should be a typo at 4) Checkpoint and > > > > >>> failover > > > > >>>>>>>> strategy > > > > >>>>>>>>> -> > > > > >>>>>>>>>>>> Mixed > > > > >>>>>>>>>>>>>> mode -> > > > > >>>>>>>>>>>>>> > > > > >>>>>>>>>>>>>> - If any task fails when isBacklog=false true, > > > > >>> this > > > > >>>>> task > > > > >>>>>>> is > > > > >>>>>>>>>>>> restarted > > > > >>>>>>>>>>>>> to > > > > >>>>>>>>>>>>>> re-process its input from the beginning. > > > > >>>>>>>>>>>>>> > > > > >>>>>>>>>>>>>> > > > > >>>>>>>>>>>>> Thank you for catching this issue. It is fixed now. > > > > >>>>>>>>>>>>> > > > > >>>>>>>>>>>>> Best, > > > > >>>>>>>>>>>>> Dong > > > > >>>>>>>>>>>>> > > > > >>>>>>>>>>>>> > > > > >>>>>>>>>>>>>> > > > > >>>>>>>>>>>>>> Best regards > > > > >>>>>>>>>>>>>> Jing > > > > >>>>>>>>>>>>>> > > > > >>>>>>>>>>>>>> > > > > >>>>>>>>>>>>>> On Thu, Jul 6, 2023 at 1:24 PM Dong Lin < > > > > >>>>>> lindon...@gmail.com > > > > >>>>>>>>>> wrote: > > > > >>>>>>>>>>>>>>> Hi Piotr, > > > > >>>>>>>>>>>>>>> > > > > >>>>>>>>>>>>>>> Thanks for your comments! Please see my reply > > > > >>> inline. > > > > >>>>>>>>>>>>>>> On Wed, Jul 5, 2023 at 11:44 PM Piotr Nowojski < > > > > >>>>>>>>>>>>> piotr.nowoj...@gmail.com > > > > >>>>>>>>>>>>>>> wrote: > > > > >>>>>>>>>>>>>>> > > > > >>>>>>>>>>>>>>>> Hi Dong, > > > > >>>>>>>>>>>>>>>> > > > > >>>>>>>>>>>>>>>> I have a couple of questions. > > > > >>>>>>>>>>>>>>>> > > > > >>>>>>>>>>>>>>>> Could you explain why those properties > > > > >>>>>>>>>>>>>>>> > > > > >>>>>>>>>>>>>>>> @Nullable private Boolean isOutputOnEOF = > > > > >>> null; > > > > >>>>>>>>>>>>>>>> @Nullable private Boolean > > > > >>> isOutputOnCheckpoint > > > > >>>> = > > > > >>>>>>> null; > > > > >>>>>>>>>>>>>>>> @Nullable private Boolean > > > > >>>>>> isInternalSorterSupported = > > > > >>>>>>>>> null; > > > > >>>>>>>>>>>>>>>> must be `@Nullable`, instead of having the > > > > >>> default > > > > >>>>>> value > > > > >>>>>>>> set > > > > >>>>>>>>> to > > > > >>>>>>>>>>>>>> `false`? > > > > >>>>>>>>>>>>>>> By initializing these private variables in > > > > >>>>>>>>>>> OperatorAttributesBuilder > > > > >>>>>>>>>>>> as > > > > >>>>>>>>>>>>>>> null, we can implement > > > > >>>>>> `OperatorAttributesBuilder#build()` > > > > >>>>>>> in > > > > >>>>>>>>>> such > > > > >>>>>>>>>>> a > > > > >>>>>>>>>>>>> way > > > > >>>>>>>>>>>>>>> that it can print DEBUG level logging to say > > > > >>>>>>>>>> "isOutputOnCheckpoint > > > > >>>>>>>>>>> is > > > > >>>>>>>>>>>>> not > > > > >>>>>>>>>>>>>>> explicitly set". This can help user/SRE debug > > > > >>>>> performance > > > > >>>>>>>>> issues > > > > >>>>>>>>>>> (or > > > > >>>>>>>>>>>>> lack > > > > >>>>>>>>>>>>>>> of the expected optimization) due to operators not > > > > >>>>>>> explicitly > > > > >>>>>>>>>>> setting > > > > >>>>>>>>>>>>> the > > > > >>>>>>>>>>>>>>> right operator attribute. > > > > >>>>>>>>>>>>>>> > > > > >>>>>>>>>>>>>>> For example, we might want a job to always use the > > > > >>>>> longer > > > > >>>>>>>>>>>> checkpointing > > > > >>>>>>>>>>>>>>> interval (i.e. > > > > >>>>>>>> execution.checkpointing.interval-during-backlog) > > > > >>>>>>>>>> if > > > > >>>>>>>>>>>> all > > > > >>>>>>>>>>>>>>> running operators have isOutputOnCheckpoint==false, > > > > >>>> and > > > > >>>>>> use > > > > >>>>>>>> the > > > > >>>>>>>>>>> short > > > > >>>>>>>>>>>>>>> checkpointing interval otherwise. If a user has > > > > >>>>>> explicitly > > > > >>>>>>>>>>> configured > > > > >>>>>>>>>>>>> the > > > > >>>>>>>>>>>>>>> execution.checkpointing.interval-during-backlog but > > > > >>>> the > > > > >>>>>>>>> two-phase > > > > >>>>>>>>>>>>> commit > > > > >>>>>>>>>>>>>>> sink library has not been upgraded to set > > > > >>>>>>>>>>> isOutputOnCheckpoint=true, > > > > >>>>>>>>>>>>> then > > > > >>>>>>>>>>>>>>> the job will end up using the long checkpointing > > > > >>>>>> interval, > > > > >>>>>>>> and > > > > >>>>>>>>> it > > > > >>>>>>>>>>>> will > > > > >>>>>>>>>>>>> be > > > > >>>>>>>>>>>>>>> useful to figure out what is going wrong in this > > > > >>> case > > > > >>>>> by > > > > >>>>>>>>> checking > > > > >>>>>>>>>>> the > > > > >>>>>>>>>>>>>> log. > > > > >>>>>>>>>>>>>>> Note that the default value of these fields of the > > > > >>>>>>>>>>> OperatorAttributes > > > > >>>>>>>>>>>>>>> instance built by OperatorAttributesBuilder will > > > > >>>> still > > > > >>>>> be > > > > >>>>>>>>> false. > > > > >>>>>>>>>>> The > > > > >>>>>>>>>>>>>>> following is mentioned in the Java doc of > > > > >>>>>>>>>>>>>>> `OperatorAttributesBuilder#build()`: > > > > >>>>>>>>>>>>>>> > > > > >>>>>>>>>>>>>>> /** > > > > >>>>>>>>>>>>>>> * If any operator attribute is null, we will log > > > > >>> it > > > > >>>>> at > > > > >>>>>>>> DEBUG > > > > >>>>>>>>>>> level > > > > >>>>>> >
