flink-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Aljoscha Krettek <aljos...@apache.org>
Subject Re: Thoughts About Streaming
Date Thu, 25 Jun 2015 09:36:41 GMT
I think I'll have to elaborate a bit so I created a proof-of-concept
implementation of my Ideas and ran some throughput measurements to
alleviate concerns about performance.

First, though, I want to highlight again why the current approach does not
work with out-of-order elements (which, again, occur constantly due to the
distributed nature of the system). This is the example I posted earlier:
https://gist.github.com/aljoscha/a367012646ab98208d27. The plan looks like
this:

+--+
| | Source
+--+
|
+-----+
| |
| +--+
| | | Identity Map
| +--+
| |
+-----+
|
+--+
| | Window
+--+
|
|
+--+
| | Sink
+--+

So all it does is pass the elements through an identity map and then merge
them again before the window operator. The source emits ascending integers
and the window operator has a custom timestamp extractor that uses the
integer itself as the timestamp and should create windows of size 4 (that
is elements with timestamp 0-3 are one window, the next are the elements
with timestamp 4-8, and so on). Since the topology basically doubles the
elements form the source I would expect to get these windows:
Window: 0, 0, 1, 1, 2, 2, 3, 3
Window: 4, 4, 6, 6, 7, 7, 8, 8

The output is this, however:
Window: 0, 1, 2, 3,
Window: 4, 0, 1, 2, 3, 4, 5, 6, 7,
Window: 8, 9, 10, 11,
Window: 12, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15,
Window: 16, 17, 18, 19,
Window: 20, 21, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23,
Window: 24, 25, 26, 27,
Window: 28, 29, 30, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31,

The reason is that the elements simply arrive out-of-order. Imagine what
would happen if the elements actually arrived with some delay from
different operations.

Now, on to the performance numbers. The proof-of-concept I created is
available here:
https://github.com/aljoscha/flink/tree/event-time-window-fn-mock. The basic
idea is that sources assign the current timestamp when emitting elements.
They also periodically emit watermarks that tell us that no elements with
an earlier timestamp will be emitted. The watermarks propagate through the
operators. The window operator looks at the timestamp of an element and
puts it into the buffer that corresponds to that window. When the window
operator receives a watermark it will look at the in-flight windows
(basically the buffers) and emit those windows where the window-end is
before the watermark.

For measuring throughput I did the following: The source emits tuples of
the form ("tuple", 1) in an infinite loop. The window operator sums up the
tuples, thereby counting how many tuples the window operator can handle in
a given time window. There are two different implementations for the
summation: 1) simply summing up the values in a mapWindow(), there you get
a List of all tuples and simple iterate over it. 2) using sum(1), which is
implemented as a reduce() (that uses the pre-reducer optimisations).

These are the performance numbers (Current is the current implementation,
Next is my proof-of-concept):

Tumbling (1 sec):
 - Current/Map: 1.6 mio
 - Current/Reduce: 2 mio
 - Next/Map: 2.2 mio
 - Next/Reduce: 4 mio

Sliding (5 sec, slide 1 sec):
 - Current/Map: ca 3 mio (fluctuates a lot)
 - Current/Reduce: No output
 - Next/Map: ca 4 mio (fluctuates)
 - Next/Reduce: 10 mio

The Next/Reduce variant can basically scale indefinitely with window size
because the internal state does not rely on the number of elements (it is
just the current sum). The pre-reducer for sliding elements cannot handle
the amount of tuples, it produces no output. For the two Map variants the
performance fluctuates because they always keep all the elements in an
internal buffer before emission, this seems to tax the garbage collector a
bit and leads to random pauses.

One thing that should be noted is that I had to disable the fake-element
emission thread, otherwise the Current versions would deadlock.

So, I started working on this because I thought that out-of-order
processing would be necessary for correctness. And it is certainly, But the
proof-of-concept also shows that performance can be greatly improved.

On Wed, 24 Jun 2015 at 09:46 Gyula Fóra <gyula.fora@gmail.com> wrote:
>
> I agree lets separate these topics from each other so we can get faster
> resolution.
>
> There is already a state discussion in the thread we started with Paris.
>
> On Wed, Jun 24, 2015 at 9:24 AM Kostas Tzoumas <ktzoumas@apache.org>
wrote:
>
> > I agree with supporting out-of-order out of the box :-), even if this
means
> > a major refactoring. This is the right time to refactor the streaming
API
> > before we pull it out of beta. I think that this is more important than
new
> > features in the streaming API, which can be prioritized once the API is
out
> > of beta (meaning, that IMO this is the right time to stall PRs until we
> > agree on the design).
> >
> > There are three sections in the document: windowing, state, and API. How
> > convoluted are those with each other? Can we separate the discussion or
do
> > we need to discuss those all together? I think part of the difficulty is
> > that we are discussing three design choices at once.
> >
> > On Tue, Jun 23, 2015 at 7:43 PM, Ted Dunning <ted.dunning@gmail.com>
> > wrote:
> >
> > > Out of order is ubiquitous in the real-world.  Typically, what
happens is
> > > that businesses will declare a maximum allowable delay for delayed
> > > transactions and will commit to results when that delay is reached.
> > > Transactions that arrive later than this cutoff are collected
specially
> > as
> > > corrections which are reported/used when possible.
> > >
> > > Clearly, ordering can also be violated during processing, but if the
data
> > > is originally out of order the situation can't be repaired by any
> > protocol
> > > fixes that prevent transactions from becoming disordered but has to
> > handled
> > > at the data level.
> > >
> > >
> > >
> > >
> > > On Tue, Jun 23, 2015 at 9:29 AM, Aljoscha Krettek <aljoscha@apache.org
>
> > > wrote:
> > >
> > > > I also don't like big changes but sometimes they are necessary. The
> > > reason
> > > > why I'm so adamant about out-of-order processing is that
out-of-order
> > > > elements are not some exception that occurs once in a while; they
occur
> > > > constantly in a distributed system. For example, in this:
> > > > https://gist.github.com/aljoscha/a367012646ab98208d27 the resulting
> > > > windows
> > > > are completely bogus because the current windowing system assumes
> > > elements
> > > > to globally arrive in order, which is simply not true. (The example
> > has a
> > > > source that generates increasing integers. Then these pass through a
> > map
> > > > and are unioned with the original DataStream before a window
operator.)
> > > > This simulates elements arriving from different operators at a
> > windowing
> > > > operator. The example is also DOP=1, I imagine this to get worse
with
> > > > higher DOP.
> > > >
> > > > What do you mean by costly? As I said, I have a proof-of-concept
> > > windowing
> > > > operator that can handle out-or-order elements. This is an example
> > using
> > > > the current Flink API:
> > > > https://gist.github.com/aljoscha/f8dce0691732e344bbe8.
> > > > (It is an infinite source of tuples and a 5 second window operator
that
> > > > counts the tuples.) The first problem is that this code deadlocks
> > because
> > > > of the thread that emits fake elements. If I disable the fake
element
> > > code
> > > > it works, but the throughput using my mockup is 4 times higher . The
> > gap
> > > > widens dramatically if the window size increases.
> > > >
> > > > So, it actually increases performance (unless I'm making a mistake
in
> > my
> > > > explorations) and can handle elements that arrive out-of-order
(which
> > > > happens basically always in a real-world windowing use-cases).
> > > >
> > > >
> > > > On Tue, 23 Jun 2015 at 12:51 Stephan Ewen <sewen@apache.org> wrote:
> > > >
> > > > > What I like a lot about Aljoscha's proposed design is that we
need no
> > > > > different code for "system time" vs. "event time". It only
differs in
> > > > where
> > > > > the timestamps are assigned.
> > > > >
> > > > > The OOP approach also gives you the semantics of total ordering
> > without
> > > > > imposing merges on the streams.
> > > > >
> > > > > On Tue, Jun 23, 2015 at 10:43 AM, Matthias J. Sax <
> > > > > mjsax@informatik.hu-berlin.de> wrote:
> > > > >
> > > > > > I agree that there should be multiple alternatives the user(!)
can
> > > > > > choose from. Partial out-of-order processing works for many/most
> > > > > > aggregates. However, if you consider Event-Pattern-Matching,
global
> > > > > > ordering in necessary (even if the performance penalty might
be
> > > high).
> > > > > >
> > > > > > I would also keep "system-time windows" as an alternative to
> > "source
> > > > > > assigned ts-windows".
> > > > > >
> > > > > > It might also be interesting to consider the following paper
for
> > > > > > overlapping windows: "Resource sharing in continuous
sliding-window
> > > > > > aggregates"
> > > > > >
> > > > > > > https://dl.acm.org/citation.cfm?id=1316720
> > > > > >
> > > > > > -Matthias
> > > > > >
> > > > > > On 06/23/2015 10:37 AM, Gyula Fóra wrote:
> > > > > > > Hey
> > > > > > >
> > > > > > > I think we should not block PRs unnecessarily if your
suggested
> > > > changes
> > > > > > > might touch them at some point.
> > > > > > >
> > > > > > > Also I still think we should not put everything in the
Datastream
> > > > > because
> > > > > > > it will be a huge mess.
> > > > > > >
> > > > > > > Also we need to agree on the out of order processing, whether
we
> > > want
> > > > > it
> > > > > > > the way you proposed it(which is quite costly). Another
> > alternative
> > > > > > > approach there which fits in the current windowing is to
filter
> > out
> > > > if
> > > > > > > order events and apply a special handling operator on them.
This
> > > > would
> > > > > be
> > > > > > > fairly lightweight.
> > > > > > >
> > > > > > > My point is that we need to consider some alternative
solutions.
> > > And
> > > > we
> > > > > > > should not block contributions along the way.
> > > > > > >
> > > > > > > Cheers
> > > > > > > Gyula
> > > > > > >
> > > > > > > On Tue, Jun 23, 2015 at 9:55 AM Aljoscha Krettek <
> > > > aljoscha@apache.org>
> > > > > > > wrote:
> > > > > > >
> > > > > > >> The reason I posted this now is that we need to think
about
the
> > > API
> > > > > and
> > > > > > >> windowing before proceeding with the PRs of Gabor (inverse
> > reduce)
> > > > and
> > > > > > >> Gyula (removal of "aggregate" functions on DataStream).
> > > > > > >>
> > > > > > >> For the windowing, I think that the current model does
not
work
> > > for
> > > > > > >> out-of-order processing. Therefore, the whole windowing
> > > > infrastructure
> > > > > > will
> > > > > > >> basically have to be redone. Meaning also that any
work on
the
> > > > > > >> pre-aggregators or optimizations that we do now becomes
useless.
> > > > > > >>
> > > > > > >> For the API, I proposed to restructure the interactions
between
> > > all
> > > > > the
> > > > > > >> different *DataStream classes and grouping/windowing.
(See
API
> > > > section
> > > > > > of
> > > > > > >> the doc I posted.)
> > > > > > >>
> > > > > > >> On Mon, 22 Jun 2015 at 21:56 Gyula Fóra <gyula.fora@gmail.com
>
> > > > wrote:
> > > > > > >>
> > > > > > >>> Hi Aljoscha,
> > > > > > >>>
> > > > > > >>> Thanks for the nice summary, this is a very good
initiative.
> > > > > > >>>
> > > > > > >>> I added some comments to the respective sections
(where I
didnt
> > > > fully
> > > > > > >> agree
> > > > > > >>> :).).
> > > > > > >>> At some point I think it would be good to have
a public
hangout
> > > > > session
> > > > > > >> on
> > > > > > >>> this, which could make a more dynamic discussion.
> > > > > > >>>
> > > > > > >>> Cheers,
> > > > > > >>> Gyula
> > > > > > >>>
> > > > > > >>> Aljoscha Krettek <aljoscha@apache.org> ezt
írta (időpont:
> > 2015.
> > > > jún.
> > > > > > >> 22.,
> > > > > > >>> H, 21:34):
> > > > > > >>>
> > > > > > >>>> Hi,
> > > > > > >>>> with people proposing changes to the streaming
part I also
> > > wanted
> > > > to
> > > > > > >>> throw
> > > > > > >>>> my hat into the ring. :D
> > > > > > >>>>
> > > > > > >>>> During the last few months, while I was getting
acquainted
> > with
> > > > the
> > > > > > >>>> streaming system, I wrote down some thoughts
I had about
how
> > > > things
> > > > > > >> could
> > > > > > >>>> be improved. Hopefully, they are in somewhat
coherent shape
> > now,
> > > > so
> > > > > > >>> please
> > > > > > >>>> have a look if you are interested in this:
> > > > > > >>>>
> > > > > > >>>>
> > > > > > >>>
> > > > > > >>
> > > > > >
> > > > >
> > > >
> > >
> >
https://docs.google.com/document/d/1rSoHyhUhm2IE30o5tkR8GEetjFvMRMNxvsCfoPsW6_4/edit?usp=sharing
> > > > > > >>>>
> > > > > > >>>> This mostly covers:
> > > > > > >>>>  - Timestamps assigned at sources
> > > > > > >>>>  - Out-of-order processing of elements in window
operators
> > > > > > >>>>  - API design
> > > > > > >>>>
> > > > > > >>>> Please let me know what you think. Comment
in the document
or
> > > here
> > > > > in
> > > > > > >> the
> > > > > > >>>> mailing list.
> > > > > > >>>>
> > > > > > >>>> I have a PR in the makings that would introduce
source
> > > timestamps
> > > > > and
> > > > > > >>>> watermarks for keeping track of them. I also
hacked a
> > > > > proof-of-concept
> > > > > > >>> of a
> > > > > > >>>> windowing system that is able to process out-of-order
elements
> > > > > using a
> > > > > > >>>> FlatMap operator. (It uses panes to perform
efficient
> > > > > > >> pre-aggregations.)
> > > > > > >>>>
> > > > > > >>>> Cheers,
> > > > > > >>>> Aljoscha
> > > > > > >>>>
> > > > > > >>>
> > > > > > >>
> > > > > > >
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message