Hi, I think the situation Vishnu raised is something that should be accounted. It can happen indeed that you want to condition what you evict from the window based on the result of the function to be applied.
My 2 cents... I would suggest adding a list for the elements of the stream where you can MARK them to be delete. Alternatively the iterator can be extended to have a function Iterator.markForEviction(int); These can be made available also in the apply function. Moreover, we can use this to extend the functionality such that you add MARKs either for eviction after the function has finished triggering or to be evicted in the next iteration. Dr. Radu Tudoran Research Engineer - Big Data Expert IT R&D Division HUAWEI TECHNOLOGIES Duesseldorf GmbH European Research Center Riesstrasse 25, 80992 München E-mail: radu.tudo...@huawei.com Mobile: +49 15209084330 Telephone: +49 891588344173 HUAWEI TECHNOLOGIES Duesseldorf GmbH Hansaallee 205, 40549 Düsseldorf, Germany, www.huawei.com Registered Office: Düsseldorf, Register Court Düsseldorf, HRB 56063, Managing Director: Bo PENG, Wanzhou MENG, Lifang CHEN Sitz der Gesellschaft: Düsseldorf, Amtsgericht Düsseldorf, HRB 56063, Geschäftsführer: Bo PENG, Wanzhou MENG, Lifang CHEN This e-mail and its attachments contain confidential information from HUAWEI, which is intended only for the person or entity whose address is listed above. Any use of the information contained herein in any way (including, but not limited to, total or partial disclosure, reproduction, or dissemination) by persons other than the intended recipient(s) is prohibited. If you receive this e-mail in error, please notify the sender by phone or email immediately and delete it! -----Original Message----- From: Vishnu Viswanath [mailto:vishnu.viswanat...@gmail.com] Sent: Thursday, July 07, 2016 1:28 AM To: Dev Subject: Re: [DISCUSS] Enhance Window Evictor in Flink Thank you Maxim and Aljoscha. Yes the beforeEvict and afterEvict should able address point 3. I have one more use case in my mind (which I might have to do in the later stages of POC). What if the `evictAfter` should behave differently based on the window function. For example. I have a window that got triggered and my evict function is being called after the apply function. In such cases I should be able to decide on what I should evict based on the window function. e.g., let the window have elements of type `case class Item(id: String, type: String)` and let the types be `type1` and `type2`. If window function is able to find a sequence : `type1 type2 type1`, then evict all elements of the type type2. or if the window function is able to find a sequence `type2 type2 type1`, then evict all elements of type type1 else don't evict any elements. Is this possible? or at least let the window function choose between two Evictor functions -(one for success case and one failure case) @Maxim: regarding the sorted window, actually I wanted my elements to be sorted but not for the eviction but while applying the window function (so thought this could be done easily). But it would be good to have the window sorted based on EventTime. Thanks and Regards, Vishnu Viswanath, On Wed, Jul 6, 2016 at 3:55 PM, Maxim <mfat...@gmail.com> wrote: > Actually for such evictor to be useful the window should be sorted by > some field, usually event time. What do you think about adding sorted > window abstraction? > > On Wed, Jul 6, 2016 at 11:36 AM, Aljoscha Krettek > <aljos...@apache.org> > wrote: > > > @Maxim: That's perfect I didn't think about using Iterator.remove() > > for that. I'll update the doc. What do you think Vishnu? This should > > also > cover > > your before/after case nicely. > > > > @Vishnu: The steps would be these: > > - Converge on a design in this discussion > > - Add a Jira issue here: > > https://issues.apache.org/jira/browse/FLINK > > - Work on the code an create a pull request on github > > > > The steps are also outlined here > > http://flink.apache.org/how-to-contribute.html and here > > http://flink.apache.org/contribute-code.html. > > > > - > > Aljoscha > > > > On Wed, 6 Jul 2016 at 19:45 Maxim <mfat...@gmail.com> wrote: > > > > > The new API forces iteration through every element of the buffer > > > even > if > > a > > > single value to be evicted. What about implementing > > > Iterator.remove() method for elements? The API would look like: > > > > > > public interface Evictor<T, W extends Window> extends Serializable > > > { > > > > > > /** > > > * Optionally evicts elements. Called before windowing function. > > > * > > > * @param elements The elements currently in the pane. Use > > > Iterator.remove to evict. > > > * @param size The current number of elements in the pane. > > > * @param window The {@link Window} > > > */ > > > void evictBefore(Iterable<T> elements, int size, EvictorContext > ctx); > > > > > > /** > > > * Optionally evicts elements. Called after windowing function. > > > * > > > * @param elements The elements currently in the pane. Use > > > Iterator.remove to evict. > > > * @param size The current number of elements in the pane. > > > * @param window The {@link Window} > > > */ > > > void evictAfter(Iterable<T> elements, int size, EvictorContext > > > ctx); } > > > > > > Such API allows to abort iteration at any point and evict elements > > > in > any > > > order. > > > > > > Thanks, > > > > > > Maxim. > > > > > > On Wed, Jul 6, 2016 at 9:04 AM, Vishnu Viswanath < > > > vishnu.viswanat...@gmail.com> wrote: > > > > > > > > Hi Aljoscha, > > > > > > > > Thanks. Yes the new interface seems to address points 1 and 2. > > > > of > > > > > > > > *1) I am having a use case where I have to create a custom > > > > Evictor > that > > > > will evict elements from the window based on the value (e.g., if > > > > I > have > > > > elements are of case class Item(id: Int, type:String) then evict > > elements > > > > that has type="a"). I believe this is not currently possible.* > > > > *2) this is somewhat related to 1) where there should be an > > > > option to > > > evict > > > > elements from anywhere in the window. not only from the > > > > beginning of > > the > > > > window. (e.g., apply the delta function to all elements and > > > > remove > all > > > > those don't pass. I checked the code and evict method just > > > > returns > the > > > > number of elements to be removed and processTriggerResult just > > > > skips > > > those > > > > many elements from the beginning. * > > > > *3) Add an option to enables the user to decide if the eviction > should > > > > happen before the apply function or after the apply function. > Currently > > > it > > > > is before the apply function, but I have a use case where I need > > > > to > > first > > > > apply the function and evict afterward.* > > > > > > > > I would be interested in contributing to the code base. Please > > > > let me > > > know > > > > the steps. > > > > > > > > Thanks and Regards, > > > > Vishnu Viswanath > > > > > > > > On Wed, Jul 6, 2016 at 11:49 AM, Aljoscha Krettek < > aljos...@apache.org > > > > > > > wrote: > > > > > > > > > Hi, > > > > > as mentioned in the thread on improving the Windowing API I > > > > > also > > have a > > > > > design doc just for improving WindowEvictors. I had this in my > > > > > head > > for > > > a > > > > > while but was hesitant to publish but since people are asking > > > > > about > > > this > > > > > now might be a good time to post it. Here's the doc: > > > > > > > > > > > > > > https://docs.google.com/document/d/1Rr7xzlItYqvFXLyyy-Yv0vvw8f29QYAjm5 > i9E4A_JlU/edit?usp=sharing > > > > > > > > > > Feedback/Suggestions are very welcome! Please let me know what > > > > > you > > > think. > > > > > > > > > > @Vishnu: Are you interested in contributing a solution for > > > > > this to > > the > > > > > Flink code base? I'd be very happy to work with you on this. > > > > > > > > > > Cheers, > > > > > Aljoscha > > > > > > > > > > P.S. I think it would be best to keep discussions to the ML > > > > > because comments on the doc will not be visible here for everyone. > > > > > > > > > > >
