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 >>>>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>> >>>>>>> >>>>>> >>>>> >>>> >>> >> >
