Hi Luke,
I guess the answer is that it depends on the state backend. If a set
operation in the state backend is available that is more efficient than
clear+append, then it would be beneficial to have a dedicated fn api
operation to allow for such optimization. That's something that needs to be
determined with a profiler :)
But the low hanging fruit is cross-bundle caching.
Thomas
On Mon, Aug 5, 2019 at 2:06 PM Lukasz Cwik <lc...@google.com> wrote:
> Thomas, why do you think a single round trip is needed?
>
> clear + append can be done blindly from the SDK side and it has total
> knowledge of the state at that point in time till the end of the bundle at
> which point you want to wait to get the cache token back from the runner
> for the append call so that for the next bundle you can reuse the state if
> the key wasn't processed elsewhere.
>
> Also, all state calls are "streamed" over gRPC so you don't need to wait
> for clear to complete before being able to send append.
>
> On Tue, Jul 30, 2019 at 12:58 AM jincheng sun <sunjincheng...@gmail.com>
> wrote:
>
>> Hi Rakesh,
>>
>> Glad to see you pointer this problem out!
>> +1 for add this implementation. Manage State by write-through-cache is
>> pretty important for Streaming job!
>>
>> Best, Jincheng
>>
>> Thomas Weise <t...@apache.org> 于2019年7月29日周一 下午8:54写道：
>>
>>> FYI a basic test appears to confirm the importance of the cross-bundle
>>> caching: I found that the throughput can be increased by playing with the
>>> bundle size in the Flink runner. Default caps at 1000 elements (or 1
>>> second). So on a high throughput stream the bundles would be capped by the
>>> count limit. Bumping the count limit increases the throughput by reducing
>>> the chatter over the state plane (more cache hits due to larger bundle).
>>>
>>> The next level of investigation would involve profiling. But just by
>>> looking at metrics, the CPU utilization on the Python worker side dropped
>>> significantly while on the Flink side it remains nearly same. There are no
>>> metrics for state operations on either side, I think it would be very
>>> helpful to get these in place also.
>>>
>>> Below the stateful processing code for reference.
>>>
>>> Thomas
>>>
>>>
>>> class StatefulFn(beam.DoFn):
>>> count_state_spec = userstate.CombiningValueStateSpec(
>>> 'count', beam.coders.IterableCoder(beam.coders.VarIntCoder()),
>>> sum)
>>> timer_spec = userstate.TimerSpec('timer',
>>> userstate.TimeDomain.WATERMARK)
>>>
>>> def process(self, kv, count=beam.DoFn.StateParam(count_state_spec),
>>> timer=beam.DoFn.TimerParam(timer_spec), window=beam.DoFn.WindowParam):
>>> count.add(1)
>>> timer_seconds = (window.end.micros // 1000000) - 1
>>> timer.set(timer_seconds)
>>>
>>> @userstate.on_timer(timer_spec)
>>> def process_timer(self,
>>> count=beam.DoFn.StateParam(count_state_spec), window=beam.DoFn.WindowParam):
>>> if count.read() == 0:
>>> logging.warning("###timer fired with count %d, window %s" %
>>> (count.read(), window))
>>>
>>>
>>>
>>> On Thu, Jul 25, 2019 at 5:09 AM Robert Bradshaw <rober...@google.com>
>>> wrote:
>>>
>>>> On Wed, Jul 24, 2019 at 6:21 AM Rakesh Kumar <rakeshku...@lyft.com>
>>>> wrote:
>>>> >
>>>> > Thanks Robert,
>>>> >
>>>> > I stumble on the jira that you have created some time ago
>>>> > https://jira.apache.org/jira/browse/BEAM-5428
>>>> >
>>>> > You also marked code where code changes are required:
>>>> >
>>>> https://github.com/apache/beam/blob/7688bcfc8ebb4bedf26c5c3b3fe0e13c0ec2aa6d/sdks/python/apache_beam/runners/worker/bundle_processor.py#L291
>>>> >
>>>> https://github.com/apache/beam/blob/7688bcfc8ebb4bedf26c5c3b3fe0e13c0ec2aa6d/sdks/python/apache_beam/runners/worker/bundle_processor.py#L349
>>>> >
>>>> https://github.com/apache/beam/blob/7688bcfc8ebb4bedf26c5c3b3fe0e13c0ec2aa6d/sdks/python/apache_beam/runners/worker/bundle_processor.py#L465
>>>> >
>>>> > I am willing to provide help to implement this. Let me know how I can
>>>> help.
>>>>
>>>> As far as I'm aware, no one is actively working on it right now.
>>>> Please feel free to assign yourself the JIRA entry and I'll be happy
>>>> to answer any questions you might have if (well probably when) these
>>>> pointers are insufficient.
>>>>
>>>> > On Tue, Jul 23, 2019 at 3:47 AM Robert Bradshaw <rober...@google.com>
>>>> wrote:
>>>> >>
>>>> >> This is documented at
>>>> >>
>>>> https://docs.google.com/document/d/1BOozW0bzBuz4oHJEuZNDOHdzaV5Y56ix58Ozrqm2jFg/edit#heading=h.7ghoih5aig5m
>>>> >> . Note that it requires participation of both the runner and the SDK
>>>> >> (though there are no correctness issues if one or the other side does
>>>> >> not understand the protocol, caching just won't be used).
>>>> >>
>>>> >> I don't think it's been implemented anywhere, but could be very
>>>> >> beneficial for performance.
>>>> >>
>>>> >> On Wed, Jul 17, 2019 at 6:00 PM Rakesh Kumar <rakeshku...@lyft.com>
>>>> wrote:
>>>> >> >
>>>> >> > I checked the python sdk[1] and it has similar implementation as
>>>> Java SDK.
>>>> >> >
>>>> >> > I would agree with Thomas. In case of high volume event stream and
>>>> bigger cluster size, network call can potentially cause a bottleneck.
>>>> >> >
>>>> >> > @Robert
>>>> >> > I am interested to see the proposal. Can you provide me the link
>>>> of the proposal?
>>>> >> >
>>>> >> > [1]:
>>>> https://github.com/apache/beam/blob/db59a3df665e094f0af17fe4d9df05fe420f3c16/sdks/python/apache_beam/transforms/userstate.py#L295
>>>> >> >
>>>> >> >
>>>> >> > On Tue, Jul 16, 2019 at 9:43 AM Thomas Weise <t...@apache.org>
>>>> wrote:
>>>> >> >>
>>>> >> >> Thanks for the pointer. For streaming, it will be important to
>>>> support caching across bundles. It appears that even the Java SDK doesn't
>>>> support that yet?
>>>> >> >>
>>>> >> >>
>>>> https://github.com/apache/beam/blob/77b295b1c2b0a206099b8f50c4d3180c248e252c/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java#L221
>>>> >> >>
>>>> >> >> Regarding clear/append: It would be nice if both could occur
>>>> within a single Fn Api roundtrip when the state is persisted.
>>>> >> >>
>>>> >> >> Thanks,
>>>> >> >> Thomas
>>>> >> >>
>>>> >> >>
>>>> >> >>
>>>> >> >> On Tue, Jul 16, 2019 at 6:58 AM Lukasz Cwik <lc...@google.com>
>>>> wrote:
>>>> >> >>>
>>>> >> >>> User state is built on top of read, append and clear and not off
>>>> a read and write paradigm to allow for blind appends.
>>>> >> >>>
>>>> >> >>> The optimization you speak of can be done completely inside the
>>>> SDK without any additional protocol being required as long as you clear the
>>>> state first and then append all your new data. The Beam Java SDK does this
>>>> for all runners when executed portably[1]. You could port the same logic to
>>>> the Beam Python SDK as well.
>>>> >> >>>
>>>> >> >>> 1:
>>>> https://github.com/apache/beam/blob/41478d00d34598e56471d99d0845ac16efa5b8ef/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/state/BagUserState.java#L84
>>>> >> >>>
>>>> >> >>> On Tue, Jul 16, 2019 at 5:54 AM Robert Bradshaw <
>>>> rober...@google.com> wrote:
>>>> >> >>>>
>>>> >> >>>> Python workers also have a per-bundle SDK-side cache. A
>>>> protocol has
>>>> >> >>>> been proposed, but hasn't yet been implemented in any SDKs or
>>>> runners.
>>>> >> >>>>
>>>> >> >>>> On Tue, Jul 16, 2019 at 6:02 AM Reuven Lax <re...@google.com>
>>>> wrote:
>>>> >> >>>> >
>>>> >> >>>> > It's runner dependent. Some runners (e.g. the Dataflow
>>>> runner) do have such a cache, though I think it's currently has a cap for
>>>> large bags.
>>>> >> >>>> >
>>>> >> >>>> > Reuven
>>>> >> >>>> >
>>>> >> >>>> > On Mon, Jul 15, 2019 at 8:48 PM Rakesh Kumar <
>>>> rakeshku...@lyft.com> wrote:
>>>> >> >>>> >>
>>>> >> >>>> >> Hi,
>>>> >> >>>> >>
>>>> >> >>>> >> I have been using python sdk for the application and also
>>>> using BagState in production. I was wondering whether state logic has any
>>>> write-through-cache implemented or not. If we are sending every read and
>>>> write request through network then it comes with a performance cost. We can
>>>> avoid network call for a read operation if we have write-through-cache.
>>>> >> >>>> >> I have superficially looked into the implementation and I
>>>> didn't see any cache implementation.
>>>> >> >>>> >>
>>>> >> >>>> >> is it possible to have this cache? would it cause any issue
>>>> if we have the caching layer?
>>>> >> >>>> >>
>>>>
>>>
