The bundles view of side inputs should never change during
processing and should have a point in time snapshot.
I was just trying to say that the cache token for side inputs
being deferred till side input request time simplified the
runners implementation since that is conclusively when the runner
would need to take a look at the side input. Putting them as part
of the ProcesBundleRequest complicates that but does make the SDK
implementation significantly simpler which is a win.
On Tue, Aug 27, 2019 at 9:14 AM Maximilian Michels
<m...@apache.org <mailto:m...@apache.org>> wrote:
Thanks for the quick response.
Just to clarify, the issue with versioning side input is also
present
when supplying the cache tokens on a request basis instead of
per
bundle. The SDK never knows when the Runner receives a new
version of
the side input. Like you pointed out, it needs to mark side
inputs as
stale and generate new cache tokens for the stale side inputs.
The difference between per-request tokens and per-bundle
tokens would be
that the side input can only change after a bundle completes
vs. during
the bundle. Side inputs are always fuzzy in that regard
because there is
no precise instance where side inputs are atomically updated,
other than
the assumption that they eventually will be updated. In that
regard
per-bundle tokens for side input seem to be fine.
All of the above is not an issue for user state, as its cache
can remain
valid for the lifetime of a Runner<=>SDK Harness connection.
A simple
solution would be to not cache side input because there are
many cases
where the caching just adds additional overhead. However, I
can also
imagine cases where side input is valid forever and caching
would be
very beneficial.
For the first version I want to focus on user state because
that's where
I see the most benefit for caching. I don't see a problem
though for the
Runner to detect new side input and reflect that in the cache
tokens
supplied for a new bundle.
-Max
On 26.08.19 22:27, Lukasz Cwik wrote:
> Your summary below makes sense to me. I can see that
recovery from
> rolling back doesn't need to be a priority and simplifies
the solution
> for user state caching down to one token.
>
> Providing cache tokens upfront does require the Runner to
know what
> "version" of everything it may supply to the SDK upfront
(instead of on
> request) which would mean that the Runner may need to have
a mapping
> from cache token to internal version identifier for things
like side
> inputs which are typically broadcast. The Runner would also
need to poll
> to see if the side input has changed in the background to
not block
> processing bundles with "stale" side input data.
>
> Ping me once you have the Runner PR updated and I'll take a
look again.
>
> On Mon, Aug 26, 2019 at 12:20 PM Maximilian Michels
<m...@apache.org <mailto:m...@apache.org>
> <mailto:m...@apache.org <mailto:m...@apache.org>>> wrote:
>
> Thank you for the summary Luke. I really appreciate the
effort you put
> into this!
>
> > Based upon your discussion you seem to want option #1
>
> I'm actually for option #2. The option to
cache/invalidate side inputs
> is important, and we should incorporate this in the
design. That's why
> option #1 is not flexible enough. However, a first
implementation could
> defer caching of side inputs.
>
> Option #3 was my initial thinking and the first version
of the PR, but I
> think we agreed that there wouldn't be much gain from
keeping a cache
> token per state id.
>
> Option #4 is what is specifically documented in the
reference doc and
> already part of the Proto, where valid tokens are
provided for each new
> bundle and also as part of the response of a
get/put/clear. We mentioned
> that the reply does not have to be waited on
synchronously (I mentioned
> it even), but it complicates the implementation. The
idea Thomas and I
> expressed was that a response is not even necessary if
we assume
> validity of the upfront provided cache tokens for the
lifetime of a
> bundle and that cache tokens will be invalidated as
soon as the Runner
> fails in any way. This is naturally the case for Flink
because it will
> simply "forget" its current cache tokens.
>
> I currently envision the following schema:
>
> Runner
> ======
>
> - Runner generates a globally unique cache token, one
for user state and
> one for each side input
>
> - The token is supplied to the SDK Harness for each
bundle request
>
> - For the lifetime of a Runner<=>SDK Harness connection
this cache token
> will not change
> - Runner will generate a new token if the
connection/key space changes
> between Runner and SDK Harness
>
>
> SDK
> ===
>
> - For each bundle the SDK worker stores the list of
valid cache tokens
> - The SDK Harness keep a global cache across all its
(local) workers
> which is a LRU cache: state_key => (cache_token, value)
> - get: Lookup cache using the valid cache token for the
state. If no
> match, then fetch from Runner and use the already
available token for
> caching
> - put: Put value in cache with a valid cache token, put
value to pending
> writes which will be flushed out latest when the bundle
ends
> - clear: same as put but clear cache
>
> It does look like this is not too far off from what you
were describing.
> The main difference is that we just work with a single
cache token. In
> my opinion we do not need the second cache token for
writes, as long as
> we ensure that we generate a new cache token if the
> bundle/checkpoint fails.
>
> I have a draft PR
> for the Runner: https://github.com/apache/beam/pull/9374
> for the SDK: https://github.com/apache/beam/pull/9418
>
> Note that the Runner PR needs to be updated to fully
reflected the above
> scheme. The SDK implementation is WIP. I want to make
sure that we
> clarify the design before this gets finalized.
>
> Thanks again for all your comments. Much appreciated!
>
> Cheers,
> Max
>
> On 26.08.19 19:58, Lukasz Cwik wrote:
> > There were originally a couple of ideas around how
caching could
> work:
> > 1) One cache token for the entire bundle that is
supplied up
> front. The
> > SDK caches everything using the given token. All
> reads/clear/append for
> > all types of state happen under this token. Anytime
a side input
> > changes, key processing partition range changes or a
bundle fails to
> > process, the runner chooses a new cache token
effectively
> invalidating
> > everything in the past>
> > 2) One cache token per type of state that is
supplied up front.
> > The SDK caches all requests for a given type using
the given cache
> > token. The runner can selectively choose which type
to keep and
> which to
> > invalidate. Bundle failure and key processing
partition changes
> > invalidate all user state, side input change
invalidates all side
> inputs.
> >
> > 3) One cache token per state id that is supplied up
front.
> > The SDK caches all requests for the given state id
using the
> given cache
> > token. The runner can selectively choose which to
invalidate and
> which
> > to keep. Bundle failure and key processing partition
changes
> invalidate
> > all user state, side input changes only invalidate
the side input
> that
> > changed.
> >
> > 4) A cache token on each read/clear/append that is
supplied on the
> > response of the call with an initial valid set that
is supplied at
> > start. The runner can selectively choose which to
keep on start.
> Bundle
> > failure allows runners to "roll back" to a known
good state by
> selecting
> > the previous valid cache token as part of the
initial set. Key
> > processing partition changes allow runners to keep
cached state that
> > hasn't changed since it can be tied to a version
number of the state
> > itself as part of the initial set. Side input
changes only invalidate
> > the side input that changed.
> >
> > Based upon your discussion you seem to want option
#1 which
> doesn't work
> > well with side inputs clearing cached state. If we
want to have user
> > state survive a changing side input, we would want
one of the other
> > options. I do agree that supplying the cache token
upfront is
> > significantly simpler. Currently the protos are
setup for #4 since it
> > was the most flexible and at the time the pros
outweighed the cons.
> >
> > I don't understand why you think you need to wait
for a response
> for the
> > append/clear to get its cache token since the only
reason you
> need the
> > cache token is that you want to use that cached data
when
> processing a
> > different bundle. I was thinking that the flow on
the SDK side
> would be
> > something like (assuming there is a global cache of
cache token
> -> (map
> > of state key -> data))
> > 1) Create a local cache of (map of state key ->
data) using the
> initial
> > set of valid cache tokens
> > 2) Make all mutations in place on local cache
without waiting for
> response.
> > 3) When response comes back, update global cache
with new cache
> token ->
> > (map of state key -> data)) (this is when the data
becomes visible to
> > other bundles that start processing)
> > 4) Before the bundle finishes processing, wait for all
> outstanding state
> > calls to finish.
> >
> > To implement caching on the runner side, you would
keep track of
> at most
> > 2 cache tokens per state key, one cache token
represents the initial
> > value when the bundle started while the second
represents the
> modified
> > state. If the bundle succeeds the runner passes in
the set of tokens
> > which represent the new state, if the bundle fails
you process
> using the
> > original ones.
> >
> > After thinking through the implementation again, we
could supply two
> > cache tokens for each state id, the first being the
set of initial
> > tokens if no writes happen while the second
represents the token
> to use
> > if the SDK changes the state. This gives us the
simplification
> where we
> > don't need to wait for the response before we update
the global cache
> > making a typical blocking cache much easier to do.
We also get the
> > benefit that runners can supply either the same
cache token for a
> state
> > id or different ones. If the runner supplies the
same one then its
> > telling the SDK to make modifications in place
without any rollback
> > (which is good on memory since we are reducing
copies of stuff) or if
> > the runner supplies two different ones then its
telling the SDK
> to keep
> > the old data around. If we went through with this
new option the SDK
> > side logic would be (assuming there is a global
cache of cache
> token ->
> > (map of state key -> data)):
> >
> > 1) Create an empty local set of state ids that are
dirty when
> starting a
> > new bundle (dirty set)
> >
> > For reads/gets:
> > 2A) If the request is a read (get), use dirty set to
choose which
> cache
> > token to lookup and use in the global cache. If the
global cache is
> > missing data issue the appropriate request providing
the result.
> >
> > For writes/appends/clear:
> > 2B) if the cache tokens are different for the state
id, add the
> state id
> > to the dirty set if it isn't there and perform the
appropriate
> > modification to convert the old cached state data to
the new
> state data
> > 3B) modify the global caches data
> > 4B) issue the request to the runner
> > 5B*) add this request to the set of requests to
block on before
> > completing the bundle.
> >
> > (* Note, there was another idea to update the
process bundle
> response to
> > contain the id of the last state request which would
allow the
> runner to
> > know when it has seen the last state request
allowing the SDK to not
> > block at all when finishing the bundle)
> >
> > On Thu, Aug 22, 2019 at 10:11 AM Maximilian Michels
> <m...@apache.org <mailto:m...@apache.org>
<mailto:m...@apache.org <mailto:m...@apache.org>>
> > <mailto:m...@apache.org <mailto:m...@apache.org>
<mailto:m...@apache.org <mailto:m...@apache.org>>>> wrote:
> >
> > Just to give a quick update here. Rakesh,
Thomas, and I had a
> discussion
> > about async writes from the Python SDK to the
Runner. Robert
> was also
> > present for some parts of the discussion.
> >
> > We concluded that blocking writes with the need
to refresh
> the cache
> > token each time are not going to provide enough
> throughput/latency.
> >
> > We figured that it will be enough to use a
single cache token per
> > Runner<=>SDK Harness connection. This cache
token will be
> provided by
> > the Runner in the ProcessBundleRequest. Writes
will not yield
> a new
> > cache token. The advantage is that we can use
one cache token
> for the
> > life time of the bundle and also across bundles,
unless the
> Runner
> > switches to a new Runner<=>SDK Harness
connection; then the
> Runner would
> > have to generate a new cache token.
> >
> > We might require additional cache tokens for the
side inputs.
> For now,
> > I'm planning to only tackle user state which
seems to be the
> area where
> > users have expressed the most need for caching.
> >
> > -Max
> >
> > On 21.08.19 20:05, Maximilian Michels wrote:
> > >> There is probably a misunderstanding here:
I'm suggesting
> to use
> > a worker ID instead of cache tokens, not
additionally.
> > >
> > > Ah! Misread that. We need a changing token to
indicate that the
> > cache is
> > > stale, e.g. checkpoint has failed / restoring
from an old
> > checkpoint. If
> > > the _Runner_ generates a new unique token/id
for workers
> which outlast
> > > the Runner, then this should work fine. I
don't think it is
> safe
> > for the
> > > worker to supply the id. The Runner should be
in control of
> cache
> > tokens
> > > to avoid invalid tokens.
> > >
> > >> In the PR the token is modified as part of
updating the state.
> > Doesn't the SDK need the new token to update
it's cache entry
> also?
> > That's where it would help the SDK to know the
new token upfront.
> > >
> > > If the state is updated in the Runner, a new
token has to be
> > generated.
> > > The old one is not valid anymore. The SDK will
use the updated
> > token to
> > > store the new value in the cache. I understand
that it would be
> > nice to
> > > know the token upfront. That could be possible
with some token
> > > generation scheme. On the other hand, writes
can be
> asynchronous and
> > > thus not block the UDF.
> > >
> > >> But I believe there is no need to change the
token in first
> > place, unless bundles for the same key (ranges)
can be
> processed by
> > different workers.
> > >
> > > That's certainly possible, e.g. two workers A
and B take turn
> > processing
> > > a certain key range, one bundle after another:
> > >
> > > You process a bundle with a token T with A,
then worker B
> takes over.
> > > Both have an entry with cache token T. So B
goes on to
> modify the
> > state
> > > and uses the same cache token T. Then A takes
over again. A
> would
> > have a
> > > stale cache entry but T would still be a valid
cache token.
> > >
> > >> Indeed the fact that Dataflow can dynamically
split and merge
> > these ranges is what makes it trickier. If Flink
does not
> > repartition the ranges, then things are much easier.
> > >
> > > Flink does not dynamically repartition key
ranges (yet). If
> it started
> > > to support that, we would invalidate the cache
tokens for
> the changed
> > > partitions.
> > >
> > >
> > > I'd suggest the following cache token
generation scheme:
> > >
> > > One cache token per key range for user state
and one cache
> token for
> > > each side input. On writes to user state or
changing side
> input, the
> > > associated cache token will be renewed.
> > >
> > > On the SDK side, it should be sufficient to
let the SDK
> > re-associate all
> > > its cached data belonging to a valid cache
token with a new
> cache
> > token
> > > returned by a successful write. This has to
happen in the
> active scope
> > > (i.e. user state, or a particular side input).
> > >
> > > If the key range changes, new cache tokens have to
> generated. This
> > > should happen automatically because the Runner
does not
> checkpoint
> > cache
> > > tokens and will generate new ones when it
restarts from an
> earlier
> > > checkpoint.
> > >
> > > The current PR needs to be changed to (1) only
keep a
> single cache
> > token
> > > per user state and key range (2) add support
for cache
> tokens for each
> > > side input.
> > >
> > > Hope that makes sense.
> > >
> > > -Max
> > >
> > > On 21.08.19 17:27, Reuven Lax wrote:
> > >>
> > >>
> > >> On Wed, Aug 21, 2019 at 2:16 AM Maximilian
Michels
> > <m...@apache.org <mailto:m...@apache.org>
<mailto:m...@apache.org <mailto:m...@apache.org>>
> <mailto:m...@apache.org <mailto:m...@apache.org>
<mailto:m...@apache.org <mailto:m...@apache.org>>>
> > >> <mailto:m...@apache.org
<mailto:m...@apache.org> <mailto:m...@apache.org
<mailto:m...@apache.org>>
> <mailto:m...@apache.org <mailto:m...@apache.org>
<mailto:m...@apache.org <mailto:m...@apache.org>>>>> wrote:
> > >>
> > >> Appreciate all your comments! Replying below.
> > >>
> > >>
> > >> @Luke:
> > >>
> > >> > Having cache tokens per key would be
very expensive
> indeed
> > and I
> > >> believe we should go with a single cache
token "per"
> bundle.
> > >>
> > >> Thanks for your comments on the PR. I was
thinking to
> propose
> > something
> > >> along this lines of having cache tokens
valid for a
> particular
> > >> checkpointing "epoch". That would require
even less token
> > renewal than
> > >> the per-bundle approach.
> > >>
> > >>
> > >> @Thomas, thanks for the input. Some remarks:
> > >>
> > >> > Wouldn't it be simpler to have the
runner just track a
> > unique ID
> > >> for each worker and use that to
communicate if the
> cache is
> > valid or
> > >> not?
> > >>
> > >> We do not need a unique id per worker. If
a cache token is
> > valid for a
> > >> particular worker, it is also valid for
another
> worker. That
> > is with the
> > >> assumption that key ranges are always
disjoint between the
> > workers.
> > >>
> > >> > * When the bundle is started, the
runner tells the
> worker
> > if the
> > >> cache has become invalid (since it knows
if another
> worker has
> > >> mutated state)
> > >>
> > >> This is simply done by not transferring
the particular
> cache
> > token. No
> > >> need to declare it invalid explicitly.
> > >>
> > >> > * When the worker sends mutation
requests to the
> runner, it
> > >> includes its own ID (or the runner
already has it as
> contextual
> > >> information). No need to wait for a response.
> > >>
> > >> Mutations of cached values can be freely
done as long
> as the
> > cache token
> > >> associated with the state is valid for a
particular
> bundle.
> > Only the
> > >> first time, the Runner needs to wait on
the response
> to store
> > the cache
> > >> token. This can also be done asynchronously.
> > >>
> > >> > * When the bundle is finished, the
runner records
> the last
> > writer
> > >> (only if a change occurred)
> > >>
> > >> I believe this is not necessary because
there will only be
> > one writer at
> > >> a time for a particular bundle and key
range, hence
> only one
> > writer
> > >> holds a valid cache token for a
particular state and
> key range.
> > >>
> > >>
> > >> @Reuven:
> > >>
> > >> > Dataflow divides the keyspace up into
lexicographic
> > ranges, and
> > >> creates a cache token per range.
> > >>
> > >> State is always processed partitioned by
the Flink workers
> > (hash-based,
> > >> not lexicopgrahical). I don't think that
matters though
> > because the key
> > >> ranges do not overlap between the
workers. Flink does
> not support
> > >> dynamically repartitioning the key
ranges. Even in case of
> > fine-grained
> > >> recovery of workers and their key ranges,
we would simply
> > generate new
> > >> cache tokens for a particular worker.
> > >>
> > >>
> > >> Dataflow's ranges are also hash based. When I
said
> lexicographical, I
> > >> meant lexicographical based on the
hexadecimal hash value.
> > >>
> > >> Indeed the fact that Dataflow can dynamically
split and
> merge these
> > >> ranges is what makes it trickier. If Flink
does not
> repartition the
> > >> ranges, then things are much easier.
> > >>
> > >>
> > >>
> > >> Thanks,
> > >> Max
> > >>
> > >> On 21.08.19 09:33, Reuven Lax wrote:
> > >> > Dataflow does something like this,
however since work is
> > >> > load balanced across workers a
per-worker id doesn't
> work
> > very well.
> > >> > Dataflow divides the keyspace up into
lexicographic
> ranges, and
> > >> creates
> > >> > a cache token per range.
> > >> >
> > >> > On Tue, Aug 20, 2019 at 8:35 PM Thomas
Weise
> > <t...@apache.org <mailto:t...@apache.org>
<mailto:t...@apache.org <mailto:t...@apache.org>>
> <mailto:t...@apache.org <mailto:t...@apache.org>
<mailto:t...@apache.org <mailto:t...@apache.org>>>
> > >> <mailto:t...@apache.org
<mailto:t...@apache.org> <mailto:t...@apache.org
<mailto:t...@apache.org>>
> <mailto:t...@apache.org <mailto:t...@apache.org>
<mailto:t...@apache.org <mailto:t...@apache.org>>>>
> > >> > <mailto:t...@apache.org
<mailto:t...@apache.org> <mailto:t...@apache.org
<mailto:t...@apache.org>>
> <mailto:t...@apache.org <mailto:t...@apache.org>
<mailto:t...@apache.org <mailto:t...@apache.org>>>
> > <mailto:t...@apache.org <mailto:t...@apache.org>
<mailto:t...@apache.org <mailto:t...@apache.org>>
> <mailto:t...@apache.org <mailto:t...@apache.org>
<mailto:t...@apache.org <mailto:t...@apache.org>>>>>> wrote:
> > >> >
> > >> > Commenting here vs. on the PR since
related to
> the overall
> > >> approach.
> > >> >
> > >> > Wouldn't it be simpler to have the
runner just
> track a
> > unique
> > >> ID for
> > >> > each worker and use that to
communicate if the
> cache is
> > valid
> > >> or not?
> > >> >
> > >> > * When the bundle is started, the
runner tells the
> > worker if the
> > >> > cache has become invalid (since it
knows if another
> > worker has
> > >> > mutated state)
> > >> > * When the worker sends mutation
requests to the
> runner, it
> > >> includes
> > >> > its own ID (or the runner already
has it as
> contextual
> > >> information).
> > >> > No need to wait for a response.
> > >> > * When the bundle is finished, the
runner
> records the
> > last writer
> > >> > (only if a change occurred)
> > >> >
> > >> > Whenever current worker ID and last
writer ID
> doesn't
> > match, cache
> > >> > is invalid.
> > >> >
> > >> > Thomas
> > >> >
> > >> >
> > >> > On Tue, Aug 20, 2019 at 11:42 AM
Lukasz Cwik
> > <lc...@google.com <mailto:lc...@google.com>
<mailto:lc...@google.com <mailto:lc...@google.com>>
> <mailto:lc...@google.com <mailto:lc...@google.com>
<mailto:lc...@google.com <mailto:lc...@google.com>>>
> > >> <mailto:lc...@google.com
<mailto:lc...@google.com> <mailto:lc...@google.com
<mailto:lc...@google.com>>
> <mailto:lc...@google.com <mailto:lc...@google.com>
<mailto:lc...@google.com <mailto:lc...@google.com>>>>
> > >> > <mailto:lc...@google.com
<mailto:lc...@google.com>
> <mailto:lc...@google.com <mailto:lc...@google.com>>
<mailto:lc...@google.com <mailto:lc...@google.com>
> <mailto:lc...@google.com <mailto:lc...@google.com>>>
> > <mailto:lc...@google.com
<mailto:lc...@google.com> <mailto:lc...@google.com
<mailto:lc...@google.com>>
> <mailto:lc...@google.com <mailto:lc...@google.com>
<mailto:lc...@google.com <mailto:lc...@google.com>>>>>> wrote:
> > >> >
> > >> > Having cache tokens per key
would be very
> expensive
> > indeed
> > >> and I
> > >> > believe we should go with a
single cache token
> > "per" bundle.
> > >> >
> > >> > On Mon, Aug 19, 2019 at 11:36
AM Maximilian
> Michels
> > >> > <m...@apache.org
<mailto:m...@apache.org> <mailto:m...@apache.org
<mailto:m...@apache.org>>
> <mailto:m...@apache.org <mailto:m...@apache.org>
<mailto:m...@apache.org <mailto:m...@apache.org>>>
> > <mailto:m...@apache.org <mailto:m...@apache.org>
<mailto:m...@apache.org <mailto:m...@apache.org>>
> <mailto:m...@apache.org <mailto:m...@apache.org>
<mailto:m...@apache.org <mailto:m...@apache.org>>>>
> > >> <mailto:m...@apache.org
<mailto:m...@apache.org> <mailto:m...@apache.org
<mailto:m...@apache.org>>
> <mailto:m...@apache.org <mailto:m...@apache.org>
<mailto:m...@apache.org <mailto:m...@apache.org>>>
> > <mailto:m...@apache.org <mailto:m...@apache.org>
<mailto:m...@apache.org <mailto:m...@apache.org>>
> <mailto:m...@apache.org <mailto:m...@apache.org>
<mailto:m...@apache.org <mailto:m...@apache.org>>>>>> wrote:
> > >> >
> > >> > Maybe a Beam Python expert
can chime in for
> > Rakesh's
> > >> question?
> > >> >
> > >> > Luke, I was assuming cache
tokens to be
> per key
> > and state
> > >> > id. During
> > >> > implementing an initial support on the
> Runner
> > side, I
> > >> > realized that we
> > >> > probably want cache tokens to only be
> per state
> > id. Note
> > >> > that if we had
> > >> > per-key cache tokens, the number of cache
> > tokens would
> > >> > approach the
> > >> > total number of keys in an
application.
> > >> >
> > >> > If anyone wants to have a
look, here is
> a first
> > version of
> > >> > the Runner
> > >> > side for cache tokens. Note
that I only
> > implemented cache
> > >> > tokens for
> > >> > BagUserState for now, but it can be easily
> > added for side
> > >> > inputs as well.
> > >> >
> > >> > https://github.com/apache/beam/pull/9374
> > >> >
> > >> > -Max
> > >> >
> > >> >
> > >>
> >
>
