Thanks to Tyler and Fabian for sharing your thoughts.
Regarding to the early/late update control of FLINK. IMO, each dynamic
table can have an EMIT config. For FLINK table-API, this can be easily
implemented in different manners, case by case. For instance, in window
aggregate, we could define "when EMIT a result" via a windowConf per each
window when we create windows. Unfortunately we do not have such
flexibility (as compared with TableAPI) in SQL query, we need find a way to
embed this EMIT config.
Regards,
Shaoxuan
On Fri, May 12, 2017 at 4:28 PM, Fabian Hueske <fhue...@gmail.com> wrote:
> 2017-05-11 7:14 GMT+02:00 Tyler Akidau <taki...@google.com.invalid>:
>
> > On Tue, May 9, 2017 at 3:06 PM Fabian Hueske <fhue...@gmail.com> wrote:
> >
> > > Hi Tyler,
> > >
> > > thank you very much for this excellent write-up and the super nice
> > > visualizations!
> > > You are discussing a lot of the things that we have been thinking about
> > as
> > > well from a different perspective.
> > > IMO, yours and the Flink model are pretty much aligned although we use
> a
> > > different terminology (which is not yet completely established). So
> there
> > > should be room for unification ;-)
> > >
> >
> > Good to hear, thanks for giving it a look. :-)
> >
> >
> > > Allow me a few words on the current state in Flink. In the upcoming
> 1.3.0
> > > release, we will have support for group window (TUMBLE, HOP, SESSION),
> > OVER
> > > RANGE/ROW window (without FOLLOWING), and non-windowed GROUP BY
> > > aggregations. The group windows are triggered by watermark and the over
> > > window and non-windowed aggregations emit for each input record
> > > (AtCount(1)). The window aggregations do not support early or late
> firing
> > > (late records are dropped), so no updates here. However, the
> non-windowed
> > > aggregates produce updates (in acc and acc/retract mode). Based on this
> > we
> > > will work on better control for late updates and early firing as well
> as
> > > joins in the next release.
> > >
> >
> > Impressive. At this rate there's a good chance we'll just be doing
> catchup
> > and thanking you for building everything. ;-) Do you have ideas for what
> > you want your early/late updates control to look like? That's one of the
> > areas I'd like to see better defined for us. And how deep are you going
> > with joins?
> >
>
> Right now (well actually I merged the change 1h ago) we are using a
> QueryConfig object to specify state retention intervals to be able to clean
> up state for inactive keys.
> A running a query looks like this:
>
> // -------
> val env = StreamExecutionEnvironment.getExecutionEnvironment
> val tEnv = TableEnvironment.getTableEnvironment(env)
>
> val qConf = tEnv.queryConfig.withIdleStateRetentionTime(Time.hours(12)) //
> state of inactive keys is kept for 12 hours
>
> val t: Table = tEnv.sql("SELECT a, COUNT(*) AS cnt FROM MyTable GROUP BY
> a")
> val stream: DataStream[(Boolean, Row)] = t.toRetractStream[Row](qConf) //
> boolean flag for acc/retract
>
> env.execute()
> // -------
>
> We plan to use the QueryConfig also to specify early/late updates. Our main
> motivation is to have a uniform and standard SQL for batch and streaming.
> Hence, we have to move the configuration out of the query. But I agree,
> that it would be very nice to be able to include it in the query. I think
> it should not be difficult to optionally support an EMIT clause as well.
>
>
> >
> > > Reading the document, I did not find any major difference in our
> > concepts.
> > > In fact, we are aiming to support the cases you are describing as well.
> > > I have a question though. Would you classify an OVER aggregation as a
> > > stream -> stream or stream -> table operation? It collects records to
> > > aggregate them, but emits one record for each input row. Depending on
> the
> > > window definition (i.e., with FOLLOWING CURRENT ROW), you can compute
> and
> > > emit the result record when the input record is received.
> > >
> >
> > I would call it a composite stream → stream operation (since SQL, like
> the
> > standard Beam/Flink APIs, is a higher level set of constructs than raw
> > streams/tables operations) consisting of a stream → table windowed
> grouping
> > followed by a table → stream triggering on every element, basically as
> you
> > described in the previous paragraph.
> >
> >
> That makes sense. Thanks :-)
>
>
> > -Tyler
> >
> >
> > >
> > > I'm looking forward to the second part.
> > >
> > > Cheers, Fabian
> > >
> > >
> > >
> > > 2017-05-09 0:34 GMT+02:00 Tyler Akidau <taki...@google.com.invalid>:
> > >
> > > > Any thoughts here Fabian? I'm planning to start sending out some more
> > > > emails towards the end of the week.
> > > >
> > > > -Tyler
> > > >
> > > >
> > > > On Wed, Apr 26, 2017 at 8:18 AM Tyler Akidau <taki...@google.com>
> > wrote:
> > > >
> > > > > No worries, thanks for the heads up. Good luck wrapping all that
> > stuff
> > > > up.
> > > > >
> > > > > -Tyler
> > > > >
> > > > > On Tue, Apr 25, 2017 at 12:07 AM Fabian Hueske <fhue...@gmail.com>
> > > > wrote:
> > > > >
> > > > >> Hi Tyler,
> > > > >>
> > > > >> thanks for pushing this effort and including the Flink list.
> > > > >> I haven't managed to read the doc yet, but just wanted to thank
> you
> > > for
> > > > >> the
> > > > >> write-up and let you know that I'm very interested in this
> > discussion.
> > > > >>
> > > > >> We are very close to the feature freeze of Flink 1.3 and I'm quite
> > > busy
> > > > >> getting as many contributions merged before the release is forked
> > off.
> > > > >> When that happened, I'll have more time to read and comment.
> > > > >>
> > > > >> Thanks,
> > > > >> Fabian
> > > > >>
> > > > >>
> > > > >> 2017-04-22 0:16 GMT+02:00 Tyler Akidau <taki...@google.com.invalid
> > >:
> > > > >>
> > > > >> > Good point, when you start talking about anything less than a
> full
> > > > join,
> > > > >> > triggers get involved to describe how one actually achieves the
> > > > desired
> > > > >> > semantics, and they may end up being tied to just one of the
> > inputs
> > > > >> (e.g.,
> > > > >> > you may only care about the watermark for one side of the join).
> > Am
> > > > >> > expecting us to address these sorts of details more precisely in
> > doc
> > > > #2.
> > > > >> >
> > > > >> > -Tyler
> > > > >> >
> > > > >> > On Fri, Apr 21, 2017 at 2:26 PM Kenneth Knowles
> > > > <k...@google.com.invalid
> > > > >> >
> > > > >> > wrote:
> > > > >> >
> > > > >> > > There's something to be said about having different triggering
> > > > >> depending
> > > > >> > on
> > > > >> > > which side of a join data comes from, perhaps?
> > > > >> > >
> > > > >> > > (delightful doc, as usual)
> > > > >> > >
> > > > >> > > Kenn
> > > > >> > >
> > > > >> > > On Fri, Apr 21, 2017 at 1:33 PM, Tyler Akidau
> > > > >> <taki...@google.com.invalid
> > > > >> > >
> > > > >> > > wrote:
> > > > >> > >
> > > > >> > > > Thanks for reading, Luke. The simple answer is that CoGBK is
> > > > >> basically
> > > > >> > > > flatten + GBK. Flatten is a non-grouping operation that
> merges
> > > the
> > > > >> > input
> > > > >> > > > streams into a single output stream. GBK then groups the
> data
> > > > within
> > > > >> > that
> > > > >> > > > single union stream as you might otherwise expect, yielding
> a
> > > > single
> > > > >> > > table.
> > > > >> > > > So I think it doesn't really impact things much. Grouping,
> > > > >> aggregation,
> > > > >> > > > window merging etc all just act upon the merged stream and
> > > > generate
> > > > >> > what
> > > > >> > > is
> > > > >> > > > effectively a merged table.
> > > > >> > > >
> > > > >> > > > -Tyler
> > > > >> > > >
> > > > >> > > > On Fri, Apr 21, 2017 at 12:36 PM Lukasz Cwik
> > > > >> <lc...@google.com.invalid
> > > > >> > >
> > > > >> > > > wrote:
> > > > >> > > >
> > > > >> > > > > The doc is a good read.
> > > > >> > > > >
> > > > >> > > > > I think you do a great job of explaining table -> stream,
> > > stream
> > > > >> ->
> > > > >> > > > stream,
> > > > >> > > > > and stream -> table when there is only one stream.
> > > > >> > > > > But when there are multiple streams reading/writing to a
> > > table,
> > > > >> how
> > > > >> > > does
> > > > >> > > > > that impact what occurs?
> > > > >> > > > > For example, with CoGBK you have multiple streams writing
> > to a
> > > > >> table,
> > > > >> > > how
> > > > >> > > > > does that impact window merging?
> > > > >> > > > >
> > > > >> > > > > On Thu, Apr 20, 2017 at 5:57 PM, Tyler Akidau
> > > > >> > > <taki...@google.com.invalid
> > > > >> > > > >
> > > > >> > > > > wrote:
> > > > >> > > > >
> > > > >> > > > > > Hello Beam, Calcite, and Flink dev lists!
> > > > >> > > > > >
> > > > >> > > > > > Apologies for the big cross post, but I thought this
> might
> > > be
> > > > >> > > something
> > > > >> > > > > all
> > > > >> > > > > > three communities would find relevant.
> > > > >> > > > > >
> > > > >> > > > > > Beam is finally making progress on a SQL DSL utilizing
> > > > Calcite,
> > > > >> > > thanks
> > > > >> > > > to
> > > > >> > > > > > Mingmin Xu. As you can imagine, we need to come to some
> > > > >> conclusion
> > > > >> > > > about
> > > > >> > > > > > how to elegantly support the full suite of streaming
> > > > >> functionality
> > > > >> > in
> > > > >> > > > the
> > > > >> > > > > > Beam model in via Calcite SQL. You folks in the Flink
> > > > community
> > > > >> > have
> > > > >> > > > been
> > > > >> > > > > > pushing on this (e.g., adding windowing constructs,
> > amongst
> > > > >> others,
> > > > >> > > > thank
> > > > >> > > > > > you! :-), but from my understanding we still don't have
> a
> > > full
> > > > >> spec
> > > > >> > > for
> > > > >> > > > > how
> > > > >> > > > > > to support robust streaming in SQL (including but not
> > > limited
> > > > >> to,
> > > > >> > > > e.g., a
> > > > >> > > > > > triggers analogue such as EMIT).
> > > > >> > > > > >
> > > > >> > > > > > I've been spending a lot of time thinking about this and
> > > have
> > > > >> some
> > > > >> > > > > opinions
> > > > >> > > > > > about how I think it should look that I've already
> written
> > > > down,
> > > > >> > so I
> > > > >> > > > > > volunteered to try to drive forward agreement on a
> general
> > > > >> > streaming
> > > > >> > > > SQL
> > > > >> > > > > > spec between our three communities (well, technically I
> > > > >> volunteered
> > > > >> > > to
> > > > >> > > > do
> > > > >> > > > > > that w/ Beam and Calcite, but I figured you Flink folks
> > > might
> > > > >> want
> > > > >> > to
> > > > >> > > > > join
> > > > >> > > > > > in since you're going that direction already anyway and
> > will
> > > > >> have
> > > > >> > > > useful
> > > > >> > > > > > insights :-).
> > > > >> > > > > >
> > > > >> > > > > > My plan was to do this by sharing two docs:
> > > > >> > > > > >
> > > > >> > > > > > 1. The Beam Model : Streams & Tables - This one is
> for
> > > > >> context,
> > > > >> > > and
> > > > >> > > > > > really only mentions SQL in passing. But it describes
> > the
> > > > >> > > > relationship
> > > > >> > > > > > between the Beam Model and the "streams & tables" way
> > of
> > > > >> > thinking,
> > > > >> > > > > which
> > > > >> > > > > > turns out to be useful in understanding what robust
> > > > >> streaming in
> > > > >> > > SQL
> > > > >> > > > > > might
> > > > >> > > > > > look like. Many of you probably already know some or
> > all
> > > of
> > > > >> > what's
> > > > >> > > > in
> > > > >> > > > > > here,
> > > > >> > > > > > but I felt it was necessary to have it all written
> down
> > > in
> > > > >> order
> > > > >> > > to
> > > > >> > > > > > justify
> > > > >> > > > > > some of the proposals I wanted to make in the second
> > doc.
> > > > >> > > > > >
> > > > >> > > > > > 2. A streaming SQL spec for Calcite - The goal for
> this
> > > doc
> > > > >> is
> > > > >> > > that
> > > > >> > > > it
> > > > >> > > > > > would become a general specification for what robust
> > > > >> streaming
> > > > >> > SQL
> > > > >> > > > in
> > > > >> > > > > > Calcite should look like. It would start out as a
> basic
> > > > >> proposal
> > > > >> > > of
> > > > >> > > > > what
> > > > >> > > > > > things *could* look like (combining both what things
> > look
> > > > >> like
> > > > >> > now
> > > > >> > > > as
> > > > >> > > > > > well
> > > > >> > > > > > as a set of proposed changes for the future), and we
> > > could
> > > > >> all
> > > > >> > > > iterate
> > > > >> > > > > > on
> > > > >> > > > > > it together until we get to something we're happy
> with.
> > > > >> > > > > >
> > > > >> > > > > > At this point, I have doc #1 ready, and it's a bit of a
> > > > monster,
> > > > >> > so I
> > > > >> > > > > > figured I'd share it and let folks hack at it with
> > comments
> > > if
> > > > >> they
> > > > >> > > > have
> > > > >> > > > > > any, while I try to get the second doc ready in the
> > > meantime.
> > > > As
> > > > >> > part
> > > > >> > > > of
> > > > >> > > > > > getting doc #2 ready, I'll be starting a separate thread
> > to
> > > > try
> > > > >> to
> > > > >> > > > gather
> > > > >> > > > > > input on what things are already in flight for streaming
> > SQL
> > > > >> across
> > > > >> > > the
> > > > >> > > > > > various communities, to make sure the proposal captures
> > > > >> everything
> > > > >> > > > that's
> > > > >> > > > > > going on as accurately as it can.
> > > > >> > > > > >
> > > > >> > > > > > If you have any questions or comments, I'm interested to
> > > hear
> > > > >> them.
> > > > >> > > > > > Otherwise, here's doc #1, "The Beam Model : Streams &
> > > Tables":
> > > > >> > > > > >
> > > > >> > > > > > http://s.apache.org/beam-streams-tables
> > > > >> > > > > >
> > > > >> > > > > > -Tyler
> > > > >> > > > > >
> > > > >> > > > >
> > > > >> > > >
> > > > >> > >
> > > > >> >
> > > > >>
> > > > >
> > > >
> > >
> >
>
