There is a lot caching that is done to minimize how much is read but for fault tolerance reasons, writes are committed to an external store. Internally in Dataflow we use state for a lot of operations to power triggers so it scales well and I wouldn't worry too much about how much state your dealing with until you notice it. This may or may not be true for other runners though.
Unless you benchmark it or you provide more concrete numbers around throughput (state reads and writes per second per key) I can't provide much more details. On Wed, Jun 7, 2017 at 11:14 AM, Josh <jof...@gmail.com> wrote: > Hi Lukasz, > > Have just given this a go with the state API and stateful DoFn on the > global window, as you suggested - it seems to work very well. > > I was just wondering how efficient it is when running on the Dataflow > runner, if for example, several elements with the same key arrive within a > few milliseconds of one another e.g. (k1, v1), (k1, v2), (k1, v3) ... and > in my stateful DoFn's processElement method I am reading and updating the > state via state.read() and state.write(...). Is it reading and writing to > an external store every time? Or is it doing all this in-memory? - I'm just > wondering how it will scale for a larger volume stream. > > Thanks, > Josh > > > > On Tue, Jun 6, 2017 at 11:18 PM, Josh <jof...@gmail.com> wrote: > >> Ok I see, thanks Lukasz. I will try this out tomorrow. >> >> Sorry for the confusing question! >> >> Josh >> >> On Tue, Jun 6, 2017 at 10:01 PM, Lukasz Cwik <lc...@google.com> wrote: >> >>> Based upon your descriptions, it seemed like you wanted limited >>> parallelism because of an external dependency. >>> >>> Your best bet would be to use the global window combined with a >>> StatefulDoFn. See this blog post (https://beam.apache.org/blog/ >>> 2017/02/13/stateful-processing.html) about the StatefulDoFn. >>> >>> You will not be able to use a different window function till after the >>> StatefulDoFn otherwise a GroupByKey may schedule your work on a different >>> machine since the windows for a key may differ. >>> >>> Source -> StatefulDoFn -> Window.into(my other window type) >>> >>> All our sources currently operate within the global window until a >>> Window.into happens. So there is no need to do Source -> >>> Window.into(GlobalWindow) -> StatefulDoFn -> Window.into(my other window >>> type) >>> >>> >>> On Tue, Jun 6, 2017 at 12:03 PM, <jof...@gmail.com> wrote: >>> >>>> Hmm ok, I don't quite get why what I want to do isn't supported in Beam >>>> ... I don't actually have a limited parallelism requirement, I just want to >>>> be able to partition my unbounded stream by a key determined from the >>>> elements, so that any two elements with the same key will be routed to the >>>> same worker. I want to do this because my DoFn keeps some in-memory cached >>>> state for each key (which I was planning to store at either DoFn or JVM >>>> level). Does this sound like a bad idea? >>>> >>>> >>>> On 6 Jun 2017, at 19:14, Lukasz Cwik <lc...@google.com> wrote: >>>> >>>> Your right, the window acts as a secondary key within GroupByKey >>>> (KeyA,Window1 != KeyA,Window2), which means that each of those two >>>> composite keys can be scheduled to execute at the same time. >>>> >>>> At this point I think you should challenge your limited parallelism >>>> requirement as you'll need to build something outside of Apache Beam to >>>> provide these parallelization limits across windows (e.g. lock within the >>>> same process when limiting yourself to a single machine, distributed lock >>>> service when dealing with multiple machines). >>>> >>>> The backlog of data is either going to grow infinitely at the >>>> GroupByKey or grow infinitely at the source if your pipeline can't keep up. >>>> It is up to the Runner to be smart and not produce a giant backlog at the >>>> GroupByKey since it knows how fast work is being completed (unfortunately I >>>> don't know if any Runner is this smart yet to push the backlog up to the >>>> source). >>>> >>>> On Tue, Jun 6, 2017 at 11:03 AM, Josh <jof...@gmail.com> wrote: >>>> >>>>> I see, thanks for the tips! >>>>> >>>>> Last question about this! How could this be adapted to work in a >>>>> unbounded/streaming job? To work in an unbounded job, I need to put a >>>>> Window.into with a trigger before GroupByKey. >>>>> I guess this would mean that the "shard gets processed by a single >>>>> thread in MyDofn" guarantee will only apply to messages within a single >>>>> window, and would not apply across windows? >>>>> If this is the case, is there a better solution? I would like to avoid >>>>> buffering data in windows, and want the shard guarantee to apply across >>>>> windows. >>>>> >>>>> >>>>> >>>>> On Tue, Jun 6, 2017 at 5:42 PM, Lukasz Cwik <lc...@google.com> wrote: >>>>> >>>>>> Your code looks like what I was describing. My only comment would be >>>>>> to use a deterministic hashing function which is stable across JVM >>>>>> versions >>>>>> and JVM instances as it will help in making your pipeline consistent >>>>>> across >>>>>> different runs/environments. >>>>>> >>>>>> Parallelizing across 8 instances instead of 4 would break the >>>>>> contract around GroupByKey (since it didn't group all the elements for a >>>>>> key correctly). Also, each element is the smallest unit of work and >>>>>> specifically in your pipeline you have chosen to reduce all your elements >>>>>> into 4 logical elements (each containing some proportion of your original >>>>>> data). >>>>>> >>>>>> On Tue, Jun 6, 2017 at 9:37 AM, Josh <jof...@gmail.com> wrote: >>>>>> >>>>>>> Thanks for the reply, Lukasz. >>>>>>> >>>>>>> >>>>>>> What I meant was that I want to shard my data by a "shard key", and >>>>>>> be sure that any two elements with the same "shard key" are processed by >>>>>>> the same thread on the same worker. (Or if that's not possible, by the >>>>>>> same >>>>>>> worker JVM with no thread guarantee would be good enough). It doesn't >>>>>>> actually matter to me whether there's 1 or 4 or 100 DoFn instances >>>>>>> processing the data. >>>>>>> >>>>>>> >>>>>>> It sounds like what you suggested will work for this, with the >>>>>>> downside of me needing to choose a number of shards/DoFns (e.g. 4). >>>>>>> >>>>>>> It seems a bit long and messy but am I right in thinking it would >>>>>>> look like this? ... >>>>>>> >>>>>>> >>>>>>> PCollection<MyElement> elements = ...; >>>>>>> >>>>>>> elements >>>>>>> >>>>>>> .apply(MapElements >>>>>>> >>>>>>> .into(TypeDescriptors.kvs(TypeDescriptors.integers(), >>>>>>> TypeDescriptor.of(MyElement.class))) >>>>>>> >>>>>>> .via((MyElement e) -> KV.of( >>>>>>> >>>>>>> e.getKey().toString().hashCode() % 4, e))) >>>>>>> >>>>>>> .apply(GroupByKey.create()) >>>>>>> >>>>>>> .apply(Partition.of(4, >>>>>>> >>>>>>> (Partition.PartitionFn<KV<Integer, Iterable<MyElement>>>) (kv, i) >>>>>>> -> kv.getKey())) >>>>>>> >>>>>>> .apply(ParDo.of(new MyDofn())); >>>>>>> >>>>>>> // Where MyDofn must be changed to handle a KV<Integer, >>>>>>> Iterable<MyElement>> as input instead of just a MyElement >>>>>>> >>>>>>> >>>>>>> I was wondering is there a guarantee that the runner won't >>>>>>> parallelise the final MyDofn across e.g. 8 instances instead of 4? If >>>>>>> there >>>>>>> are two input elements with the same key are they actually guaranteed >>>>>>> to be >>>>>>> processed on the same instance? >>>>>>> >>>>>>> >>>>>>> Thanks, >>>>>>> >>>>>>> Josh >>>>>>> >>>>>>> >>>>>>> >>>>>>> >>>>>>> On Tue, Jun 6, 2017 at 4:51 PM, Lukasz Cwik <lc...@google.com> >>>>>>> wrote: >>>>>>> >>>>>>>> I think this is what your asking for but your statement about 4 >>>>>>>> instances is unclear as to whether that is 4 copies of the same DoFn >>>>>>>> or 4 >>>>>>>> completely different DoFns. Also its unclear what you mean by >>>>>>>> instance/thread, I'm assuming that you want at most 4 instances of a >>>>>>>> DoFn >>>>>>>> each being processed by a single thread. >>>>>>>> >>>>>>>> This is a bad idea because you limit your parallelism but this is >>>>>>>> similar to what the default file sharding logic does. In Apache Beam >>>>>>>> the >>>>>>>> smallest unit of output for a GroupByKey is a single key+iterable >>>>>>>> pair. We >>>>>>>> exploit this by assigning all our values to a fixed number of keys and >>>>>>>> then >>>>>>>> performing a GroupByKey. This is the same trick that powers the file >>>>>>>> sharding logic in AvroIO/TextIO/... >>>>>>>> >>>>>>>> Your pipeline would look like (fixed width font diagram): >>>>>>>> your data -> apply shard key -> GroupByKey -> >>>>>>>> partition by key -> your dofn #1 >>>>>>>> >>>>>>>> \> your dofn #2 >>>>>>>> >>>>>>>> \> ... >>>>>>>> a / b / c / d -> 1,a / 2,b / 1,c / 2,d -> 1,[a,c] / 2,[b,d] -> ??? >>>>>>>> >>>>>>>> This is not exactly the same as processing a single DoFn >>>>>>>> instance/thread because it relies on the Runner to be able to schedule >>>>>>>> each >>>>>>>> key to be processed on a different machine. For example a Runner may >>>>>>>> choose >>>>>>>> to process value 1,[a,c] and 2,[b,d] sequentially on the same machine >>>>>>>> or >>>>>>>> may choose to distribute them. >>>>>>>> >>>>>>>> >>>>>>>> >>>>>>>> On Tue, Jun 6, 2017 at 8:13 AM, Josh <jof...@gmail.com> wrote: >>>>>>>> >>>>>>>>> Hey Lukasz, >>>>>>>>> >>>>>>>>> I have a follow up question about this - >>>>>>>>> >>>>>>>>> What if I want to do something very similar, but instead of with 4 >>>>>>>>> instances of AvroIO following the partition transform, I want 4 >>>>>>>>> instances >>>>>>>>> of a DoFn that I've written. I want to ensure that each partition is >>>>>>>>> processed by a single DoFn instance/thread. Is this possible with >>>>>>>>> Beam? >>>>>>>>> >>>>>>>>> Thanks, >>>>>>>>> Josh >>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> On Wed, May 24, 2017 at 6:15 PM, Josh <jof...@gmail.com> wrote: >>>>>>>>> >>>>>>>>>> Ahh I see - Ok I'll try out this solution then. Thanks Lukasz! >>>>>>>>>> >>>>>>>>>> On Wed, May 24, 2017 at 5:20 PM, Lukasz Cwik <lc...@google.com> >>>>>>>>>> wrote: >>>>>>>>>> >>>>>>>>>>> Google Cloud Dataflow won't override your setting. The dynamic >>>>>>>>>>> sharding occurs if you don't explicitly set a numShard value. >>>>>>>>>>> >>>>>>>>>>> On Wed, May 24, 2017 at 9:14 AM, Josh <jof...@gmail.com> wrote: >>>>>>>>>>> >>>>>>>>>>>> Hi Lukasz, >>>>>>>>>>>> >>>>>>>>>>>> Thanks for the example. That sounds like a nice solution - >>>>>>>>>>>> I am running on Dataflow though, which dynamically sets >>>>>>>>>>>> numShards - so if I set numShards to 1 on each of those AvroIO >>>>>>>>>>>> writers, I >>>>>>>>>>>> can't be sure that Dataflow isn't going to override my setting >>>>>>>>>>>> right? I >>>>>>>>>>>> guess this should work fine as long as I partition my stream into >>>>>>>>>>>> a large >>>>>>>>>>>> enough number of partitions so that Dataflow won't override >>>>>>>>>>>> numShards. >>>>>>>>>>>> >>>>>>>>>>>> Josh >>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>>> On Wed, May 24, 2017 at 4:10 PM, Lukasz Cwik <lc...@google.com> >>>>>>>>>>>> wrote: >>>>>>>>>>>> >>>>>>>>>>>>> Since your using a small number of shards, add a Partition >>>>>>>>>>>>> transform which uses a deterministic hash of the key to choose >>>>>>>>>>>>> one of 4 >>>>>>>>>>>>> partitions. Write each partition with a single shard. >>>>>>>>>>>>> >>>>>>>>>>>>> (Fixed width diagram below) >>>>>>>>>>>>> Pipeline -> AvroIO(numShards = 4) >>>>>>>>>>>>> Becomes: >>>>>>>>>>>>> Pipeline -> Partition --> AvroIO(numShards = 1) >>>>>>>>>>>>> |-> AvroIO(numShards = 1) >>>>>>>>>>>>> |-> AvroIO(numShards = 1) >>>>>>>>>>>>> \-> AvroIO(numShards = 1) >>>>>>>>>>>>> >>>>>>>>>>>>> On Wed, May 24, 2017 at 1:05 AM, Josh <jof...@gmail.com> >>>>>>>>>>>>> wrote: >>>>>>>>>>>>> >>>>>>>>>>>>>> Hi, >>>>>>>>>>>>>> >>>>>>>>>>>>>> I am using a FileBasedSink (AvroIO.write) on an unbounded >>>>>>>>>>>>>> stream (withWindowedWrites, hourly windows, numShards=4). >>>>>>>>>>>>>> >>>>>>>>>>>>>> I would like to partition the stream by some key in the >>>>>>>>>>>>>> element, so that all elements with the same key will get >>>>>>>>>>>>>> processed by the >>>>>>>>>>>>>> same shard writer, and therefore written to the same file. Is >>>>>>>>>>>>>> there a way >>>>>>>>>>>>>> to do this? Note that in my stream the number of keys is very >>>>>>>>>>>>>> large (most >>>>>>>>>>>>>> elements have a unique key, while a few elements share a key). >>>>>>>>>>>>>> >>>>>>>>>>>>>> Thanks, >>>>>>>>>>>>>> Josh >>>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>> >>>>>>>>> >>>>>>>> >>>>>>> >>>>>> >>>>> >>>> >>> >> >
