Hi Jay,
Thanks for the advise, I would like to list down the use cases as
per your suggestion. But it seems I don't have write permission to the
Apache Kafka Confluent Space. Whom shall I request for it?
Regarding your last question. We are using a patch in our production system
which does exactly this.
We window by the event time, but trigger punctuate in <punctuate interval>
duration of system time, in the absence of an event crossing the punctuate
event time.
We are using Kafka Stream for our Audit Trail, where we need to output the
event counts on each topic on each cluster aggregated over a 1 minute
window. We have to use event time to be able to cross check the counts. But
we need to trigger punctuate [aggregate event pushes] by system time in the
absence of events. Otherwise the event counts for unexpired windows would
be 0 which is bad.
"Maybe a hybrid solution works: I window by event time but trigger results
by system time for windows that have updated? Not really sure the details
of making that work. Does that work? Are there concrete examples where you
actually want the current behavior?"
--
With Regards,
Arun Mathew
Yahoo! JAPAN Corporation
On Wed, Apr 5, 2017 at 8:48 PM, Tianji Li <skyah...@gmail.com> wrote:
> Hi Jay,
>
> The hybrid solution is exactly what I expect and need for our use cases
> when dealing with telecom data.
>
> Thanks
> Tianji
>
> On Wed, Apr 5, 2017 at 12:01 AM, Jay Kreps <j...@confluent.io> wrote:
>
> > Hey guys,
> >
> > One thing I've always found super important for this kind of design work
> is
> > to do a really good job of cataloging the landscape of use cases and how
> > prevalent each one is. By that I mean not just listing lots of uses, but
> > also grouping them into categories that functionally need the same thing.
> > In the absence of this it is very hard to reason about design proposals.
> > From the proposals so far I think we have a lot of discussion around
> > possible apis, but less around what the user needs for different use
> cases
> > and how they would implement that using the api.
> >
> > Here is an example:
> > You aggregate click and impression data for a reddit like site. Every ten
> > minutes you want to output a ranked list of the top 10 articles ranked by
> > clicks/impressions for each geographical area. I want to be able run this
> > in steady state as well as rerun to regenerate results (or catch up if it
> > crashes).
> >
> > There are a couple of tricky things that seem to make this hard with
> either
> > of the options proposed:
> > 1. If I emit this data using event time I have the problem described
> where
> > a geographical region with no new clicks or impressions will fail to
> output
> > results.
> > 2. If I emit this data using system time I have the problem that when
> > reprocessing data my window may not be ten minutes but 10 hours if my
> > processing is very fast so it dramatically changes the output.
> >
> > Maybe a hybrid solution works: I window by event time but trigger results
> > by system time for windows that have updated? Not really sure the details
> > of making that work. Does that work? Are there concrete examples where
> you
> > actually want the current behavior?
> >
> > -Jay
> >
> >
> > On Tue, Apr 4, 2017 at 8:32 PM, Arun Mathew <arunmathe...@gmail.com>
> > wrote:
> >
> > > Hi All,
> > >
> > > Thanks for the KIP. We were also in need of a mechanism to trigger
> > > punctuate in the absence of events.
> > >
> > > As I described in [
> > > https://issues.apache.org/jira/browse/KAFKA-3514?
> > > focusedCommentId=15926036&page=com.atlassian.jira.
> > > plugin.system.issuetabpanels:comment-tabpanel#comment-15926036
> > > ],
> > >
> > > - Our approached involved using the event time by default.
> > > - The method to check if there is any punctuate ready in the
> > > PunctuationQueue is triggered via the any event received by the
> stream
> > > tread, or at the polling intervals in the absence of any events.
> > > - When we create Punctuate objects (which contains the next event
> time
> > > for punctuation and interval), we also record the creation time
> > (system
> > > time).
> > > - While checking for maturity of Punctuate Schedule by
> mayBePunctuate
> > > method, we also check if the system clock has elapsed the punctuate
> > > interval since the schedule creation time.
> > > - In the absence of any event, or in the absence of any event for
> one
> > > topic in the partition group assigned to the stream task, the system
> > > time
> > > will elapse the interval and we trigger a punctuate using the
> expected
> > > punctuation event time.
> > > - we then create the next punctuation schedule as punctuation event
> > time
> > > + punctuation interval, [again recording the system time of creation
> > of
> > > the
> > > schedule].
> > >
> > > We call this a Hybrid Punctuate. Of course, this approach has pros and
> > > cons.
> > > Pros
> > >
> > > - Punctuates will happen in <punctuate interval> time duration at
> max
> > in
> > > terms of system time.
> > > - The semantics as a whole continues to revolve around event time.
> > > - We can use the old data [old timestamps] to rerun any experiments
> or
> > > tests.
> > >
> > > Cons
> > >
> > > - In case the <punctuate interval> is not a time duration [say
> > logical
> > > time/event count], then the approach might not be meaningful.
> > > - In case there is a case where we have to wait for an actual event
> > from
> > > a low event rate partition in the partition group, this approach
> will
> > > jump
> > > the gun.
> > > - in case the event processing cannot catch up with the event rate
> and
> > > the expected timestamp events gets queued for long time, this
> approach
> > > might jump the gun.
> > >
> > > I believe the above approach and discussion goes close to the approach
> A.
> > >
> > > -----------
> > >
> > > I like the idea of having an even count based punctuate.
> > >
> > > -----------
> > >
> > > I agree with the discussion around approach C, that we should provide
> the
> > > user with the option to choose system time or event time based
> > punctuates.
> > > But I believe that the user predominantly wants to use event time while
> > not
> > > missing out on regular punctuates due to event delays or event
> absences.
> > > Hence a complex punctuate option as Matthias mentioned (quoted below)
> > would
> > > be most apt.
> > >
> > > "- We might want to add "complex" schedules later on (like, punctuate
> on
> > > every 10 seconds event-time or 60 seconds system-time whatever comes
> > > first)."
> > >
> > > -----------
> > >
> > > I think I read somewhere that Kafka Streams started with System Time as
> > the
> > > punctuation standard, but was later changed to Event Time. I guess
> there
> > > would be some good reason behind it. As Kafka Streams want to evolve
> more
> > > on the Stream Processing front, I believe the emphasis on event time
> > would
> > > remain quite strong.
> > >
> > >
> > > With Regards,
> > >
> > > Arun Mathew
> > > Yahoo! JAPAN Corporation, Tokyo
> > >
> > >
> > > On Wed, Apr 5, 2017 at 3:53 AM, Thomas Becker <tobec...@tivo.com>
> wrote:
> > >
> > > > Yeah I like PuncutationType much better; I just threw Time out there
> > > > more as a strawman than an actual suggestion ;) I still think it's
> > > > worth considering what this buys us over an additional callback. I
> > > > foresee a number of punctuate implementations following this pattern:
> > > >
> > > > public void punctuate(PunctuationType type) {
> > > > switch (type) {
> > > > case EVENT_TIME:
> > > > methodA();
> > > > break;
> > > > case SYSTEM_TIME:
> > > > methodB();
> > > > break;
> > > > }
> > > > }
> > > >
> > > > I guess one advantage of this approach is we could add additional
> > > > punctuation types later in a backwards compatible way (like event
> count
> > > > as you mentioned).
> > > >
> > > > -Tommy
> > > >
> > > >
> > > > On Tue, 2017-04-04 at 11:10 -0700, Matthias J. Sax wrote:
> > > > > That sounds promising.
> > > > >
> > > > > I am just wondering if `Time` is the best name. Maybe we want to
> add
> > > > > other non-time based punctuations at some point later. I would
> > > > > suggest
> > > > >
> > > > > enum PunctuationType {
> > > > > EVENT_TIME,
> > > > > SYSTEM_TIME,
> > > > > }
> > > > >
> > > > > or similar. Just to keep the door open -- it's easier to add new
> > > > > stuff
> > > > > if the name is more generic.
> > > > >
> > > > >
> > > > > -Matthias
> > > > >
> > > > >
> > > > > On 4/4/17 5:30 AM, Thomas Becker wrote:
> > > > > >
> > > > > > I agree that the framework providing and managing the notion of
> > > > > > stream
> > > > > > time is valuable and not something we would want to delegate to
> the
> > > > > > tasks. I'm not entirely convinced that a separate callback
> (option
> > > > > > C)
> > > > > > is that messy (it could just be a default method with an empty
> > > > > > implementation), but if we wanted a single API to handle both
> > > > > > cases,
> > > > > > how about something like the following?
> > > > > >
> > > > > > enum Time {
> > > > > > STREAM,
> > > > > > CLOCK
> > > > > > }
> > > > > >
> > > > > > Then on ProcessorContext:
> > > > > > context.schedule(Time time, long interval) // We could allow
> this
> > > > > > to
> > > > > > be called once for each value of time to mix approaches.
> > > > > >
> > > > > > Then the Processor API becomes:
> > > > > > punctuate(Time time) // time here denotes which schedule resulted
> > > > > > in
> > > > > > this call.
> > > > > >
> > > > > > Thoughts?
> > > > > >
> > > > > >
> > > > > > On Mon, 2017-04-03 at 22:44 -0700, Matthias J. Sax wrote:
> > > > > > >
> > > > > > > Thanks a lot for the KIP Michal,
> > > > > > >
> > > > > > > I was thinking about the four options you proposed in more
> > > > > > > details
> > > > > > > and
> > > > > > > this are my thoughts:
> > > > > > >
> > > > > > > (A) You argue, that users can still "punctuate" on event-time
> via
> > > > > > > process(), but I am not sure if this is possible. Note, that
> > > > > > > users
> > > > > > > only
> > > > > > > get record timestamps via context.timestamp(). Thus, users
> would
> > > > > > > need
> > > > > > > to
> > > > > > > track the time progress per partition (based on the partitions
> > > > > > > they
> > > > > > > obverse via context.partition(). (This alone puts a huge burden
> > > > > > > on
> > > > > > > the
> > > > > > > user by itself.) However, users are not notified at startup
> what
> > > > > > > partitions are assigned, and user are not notified when
> > > > > > > partitions
> > > > > > > get
> > > > > > > revoked. Because this information is not available, it's not
> > > > > > > possible
> > > > > > > to
> > > > > > > "manually advance" stream-time, and thus event-time punctuation
> > > > > > > within
> > > > > > > process() seems not to be possible -- or do you see a way to
> get
> > > > > > > it
> > > > > > > done? And even if, it might still be too clumsy to use.
> > > > > > >
> > > > > > > (B) This does not allow to mix both approaches, thus limiting
> > > > > > > what
> > > > > > > users
> > > > > > > can do.
> > > > > > >
> > > > > > > (C) This should give all flexibility we need. However, just
> > > > > > > adding
> > > > > > > one
> > > > > > > more method seems to be a solution that is too simple (cf my
> > > > > > > comments
> > > > > > > below).
> > > > > > >
> > > > > > > (D) This might be hard to use. Also, I am not sure how a user
> > > > > > > could
> > > > > > > enable system-time and event-time punctuation in parallel.
> > > > > > >
> > > > > > >
> > > > > > >
> > > > > > > Overall options (C) seems to be the most promising approach to
> > > > > > > me.
> > > > > > > Because I also favor a clean API, we might keep current
> > > > > > > punctuate()
> > > > > > > as-is, but deprecate it -- so we can remove it at some later
> > > > > > > point
> > > > > > > when
> > > > > > > people use the "new punctuate API".
> > > > > > >
> > > > > > >
> > > > > > > Couple of follow up questions:
> > > > > > >
> > > > > > > - I am wondering, if we should have two callback methods or
> just
> > > > > > > one
> > > > > > > (ie, a unified for system and event time punctuation or one for
> > > > > > > each?).
> > > > > > >
> > > > > > > - If we have one, how can the user figure out, which condition
> > > > > > > did
> > > > > > > trigger?
> > > > > > >
> > > > > > > - How would the API look like, for registering different
> > > > > > > punctuate
> > > > > > > schedules? The "type" must be somehow defined?
> > > > > > >
> > > > > > > - We might want to add "complex" schedules later on (like,
> > > > > > > punctuate
> > > > > > > on
> > > > > > > every 10 seconds event-time or 60 seconds system-time whatever
> > > > > > > comes
> > > > > > > first). I don't say we should add this right away, but we might
> > > > > > > want
> > > > > > > to
> > > > > > > define the API in a way, that it allows extensions like this
> > > > > > > later
> > > > > > > on,
> > > > > > > without redesigning the API (ie, the API should be designed
> > > > > > > extensible)
> > > > > > >
> > > > > > > - Did you ever consider count-based punctuation?
> > > > > > >
> > > > > > >
> > > > > > > I understand, that you would like to solve a simple problem,
> but
> > > > > > > we
> > > > > > > learned from the past, that just "adding some API" quickly
> leads
> > > > > > > to a
> > > > > > > not very well defined API that needs time consuming clean up
> > > > > > > later on
> > > > > > > via other KIPs. Thus, I would prefer to get a holistic
> > > > > > > punctuation
> > > > > > > KIP
> > > > > > > with this from the beginning on to avoid later painful
> redesign.
> > > > > > >
> > > > > > >
> > > > > > >
> > > > > > > -Matthias
> > > > > > >
> > > > > > >
> > > > > > >
> > > > > > > On 4/3/17 11:58 AM, Michal Borowiecki wrote:
> > > > > > > >
> > > > > > > >
> > > > > > > > Thanks Thomas,
> > > > > > > >
> > > > > > > > I'm also wary of changing the existing semantics of
> punctuate,
> > > > > > > > for
> > > > > > > > backward compatibility reasons, although I like the
> conceptual
> > > > > > > > simplicity of that option.
> > > > > > > >
> > > > > > > > Adding a new method to me feels safer but, in a way, uglier.
> I
> > > > > > > > added
> > > > > > > > this to the KIP now as option (C).
> > > > > > > >
> > > > > > > > The TimestampExtractor mechanism is actually more flexible,
> as
> > > > > > > > it
> > > > > > > > allows
> > > > > > > > you to return any value, you're not limited to event time or
> > > > > > > > system
> > > > > > > > time
> > > > > > > > (although I don't see an actual use case where you might need
> > > > > > > > anything
> > > > > > > > else then those two). Hence I also proposed the option to
> allow
> > > > > > > > users
> > > > > > > > to, effectively, decide what "stream time" is for them given
> > > > > > > > the
> > > > > > > > presence or absence of messages, much like they can decide
> what
> > > > > > > > msg
> > > > > > > > time
> > > > > > > > means for them using the TimestampExtractor. What do you
> think
> > > > > > > > about
> > > > > > > > that? This is probably most flexible but also most
> complicated.
> > > > > > > >
> > > > > > > > All comments appreciated.
> > > > > > > >
> > > > > > > > Cheers,
> > > > > > > >
> > > > > > > > Michal
> > > > > > > >
> > > > > > > >
> > > > > > > > On 03/04/17 19:23, Thomas Becker wrote:
> > > > > > > > >
> > > > > > > > >
> > > > > > > > > Although I fully agree we need a way to trigger periodic
> > > > > > > > > processing
> > > > > > > > > that is independent from whether and when messages arrive,
> > > > > > > > > I'm
> > > > > > > > > not sure
> > > > > > > > > I like the idea of changing the existing semantics across
> the
> > > > > > > > > board.
> > > > > > > > > What if we added an additional callback to Processor that
> can
> > > > > > > > > be
> > > > > > > > > scheduled similarly to punctuate() but was always called at
> > > > > > > > > fixed, wall
> > > > > > > > > clock based intervals? This way you wouldn't have to give
> up
> > > > > > > > > the
> > > > > > > > > notion
> > > > > > > > > of stream time to be able to do periodic processing.
> > > > > > > > >
> > > > > > > > > On Mon, 2017-04-03 at 10:34 +0100, Michal Borowiecki wrote:
> > > > > > > > > >
> > > > > > > > > >
> > > > > > > > > > Hi all,
> > > > > > > > > >
> > > > > > > > > > I have created a draft for KIP-138: Change punctuate
> > > > > > > > > > semantics
> > > > > > > > > > <https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> > 138%
> > > > > > > > > > 3A+C
> > > > > > > > > > hange+
> > > > > > > > > > punctuate+semantics>
> > > > > > > > > > .
> > > > > > > > > >
> > > > > > > > > > Appreciating there can be different views on system-time
> vs
> > > > > > > > > > event-
> > > > > > > > > > time
> > > > > > > > > > semantics for punctuation depending on use-case and the
> > > > > > > > > > importance of
> > > > > > > > > > backwards compatibility of any such change, I've left it
> > > > > > > > > > quite
> > > > > > > > > > open
> > > > > > > > > > and
> > > > > > > > > > hope to fill in more info as the discussion progresses.
> > > > > > > > > >
> > > > > > > > > > Thanks,
> > > > > > > > > > Michal
> > > > > > --
> > > > > >
> > > > > >
> > > > > > Tommy Becker
> > > > > >
> > > > > > Senior Software Engineer
> > > > > >
> > > > > > O +1 919.460.4747
> > > > > >
> > > > > > tivo.com
> > > > > >
> > > > > >
> > > > > > ________________________________
> > > > > >
> > > > > > This email and any attachments may contain confidential and
> > > > > > privileged material for the sole use of the intended recipient.
> Any
> > > > > > review, copying, or distribution of this email (or any
> attachments)
> > > > > > by others is prohibited. If you are not the intended recipient,
> > > > > > please contact the sender immediately and permanently delete this
> > > > > > email and any attachments. No employee or agent of TiVo Inc. is
> > > > > > authorized to conclude any binding agreement on behalf of TiVo
> Inc.
> > > > > > by email. Binding agreements with TiVo Inc. may only be made by a
> > > > > > signed written agreement.
> > > > > >
> > > > --
> > > >
> > > >
> > > > Tommy Becker
> > > >
> > > > Senior Software Engineer
> > > >
> > > > O +1 919.460.4747
> > > >
> > > > tivo.com
> > > >
> > > >
> > > > ________________________________
> > > >
> > > > This email and any attachments may contain confidential and
> privileged
> > > > material for the sole use of the intended recipient. Any review,
> > copying,
> > > > or distribution of this email (or any attachments) by others is
> > > prohibited.
> > > > If you are not the intended recipient, please contact the sender
> > > > immediately and permanently delete this email and any attachments. No
> > > > employee or agent of TiVo Inc. is authorized to conclude any binding
> > > > agreement on behalf of TiVo Inc. by email. Binding agreements with
> TiVo
> > > > Inc. may only be made by a signed written agreement.
> > > >
> > >
> >
>
