Great, let me know when to take another look at the PR! Reuven
On Wed, May 1, 2019 at 6:47 AM Jozef Vilcek <jozo.vil...@gmail.com> wrote: > That coder is added extra as a re-map stage from "original" key to new > ShardAwareKey ... But pipeline might get broken I guess. > Very fair point. I am having a second thought pass over this and will try > to simplify it much more > > On Wed, May 1, 2019 at 2:12 PM Reuven Lax <re...@google.com> wrote: > >> I haven't looked at the PR in depth yet, but it appears that someone >> running a pipeline today who then tries to update post this PR will have >> the coder change to DefaultShardKeyCoder, even if they haven't picked any >> custom function. Is that correct, or am I misreading things? >> >> Reuven >> >> On Tue, Apr 30, 2019 at 8:42 AM Jozef Vilcek <jozo.vil...@gmail.com> >> wrote: >> >>> Hm, what would be the scenario? Have version A running with original >>> random sharding and then start version B where I change sharding to some >>> custom function? >>> So I have to enable the pipeline to digest old keys from GBK restored >>> state and also work with new keys produced to GBK going forward? >>> >>> On Tue, Apr 30, 2019 at 5:32 PM Reuven Lax <re...@google.com> wrote: >>> >>>> Initial thought on PR: we usually try to limit changing coders in these >>>> types of transforms to better support runners that allow in-place updates >>>> of pipelines. Can this be done without changing the coder? >>>> >>>> On Tue, Apr 30, 2019 at 8:21 AM Jozef Vilcek <jozo.vil...@gmail.com> >>>> wrote: >>>> >>>>> I have created a PR for enhancing WriteFiles for custom sharding >>>>> function. >>>>> https://github.com/apache/beam/pull/8438 >>>>> >>>>> If this sort of change looks good, then next step would be to use in >>>>> in Flink runner transform override. Let me know what do you think >>>>> >>>>> On Fri, Apr 26, 2019 at 9:24 AM Jozef Vilcek <jozo.vil...@gmail.com> >>>>> wrote: >>>>> >>>>>> I guess it is fine to enable shardingFn control only on WriteFiles >>>>>> level rather than FileIO. On WriteFiles it can be manipulated in >>>>>> PTransformOverride by runner. >>>>>> >>>>>> On Thu, Apr 25, 2019 at 6:17 PM Reuven Lax <re...@google.com> wrote: >>>>>> >>>>>>> Yes, a hook would have to be added to allow specifying a different >>>>>>> function for choosing the shard number (I assume the problem is that >>>>>>> there >>>>>>> are cases where the current random assignment is not good?). However >>>>>>> this >>>>>>> can be set using PTransformOverride, we ideally shouldn't force the >>>>>>> user to >>>>>>> know details of the runner when writing their code. >>>>>>> >>>>>>> On Thu, Apr 25, 2019 at 7:52 AM Maximilian Michels <m...@apache.org> >>>>>>> wrote: >>>>>>> >>>>>>>> Reuven is talking about PTransformOverride, e.g. >>>>>>>> FlinkTransformOverrides. We already use this to determine the >>>>>>>> number of >>>>>>>> shards in case of Runner-determined sharding. >>>>>>>> >>>>>>>> Not sure if that would work for Jozef's case because setting the >>>>>>>> number >>>>>>>> of shards is not enough. We want to set the shard key directly and >>>>>>>> that >>>>>>>> logic is buried inside WriteFiles. >>>>>>>> >>>>>>>> -Max >>>>>>>> >>>>>>>> On 25.04.19 16:30, Reuven Lax wrote: >>>>>>>> > Actually the runner is free to perform surgery on the graph. The >>>>>>>> > FlinkRunner can insert a custom function to determine the >>>>>>>> sharding keys. >>>>>>>> > >>>>>>>> > On Thu, Apr 25, 2019 at 7:28 AM Jozef Vilcek < >>>>>>>> jozo.vil...@gmail.com >>>>>>>> > <mailto:jozo.vil...@gmail.com>> wrote: >>>>>>>> > >>>>>>>> > Right now, sharding can be specified only via target >>>>>>>> `shardCount`, >>>>>>>> > be it user or runner. Next to configurable shardCount, I am >>>>>>>> > proposing to be able to pass also a function which will allow >>>>>>>> to the >>>>>>>> > user (or runner) control how is shard determined and what key >>>>>>>> will >>>>>>>> > be used to represent it >>>>>>>> > >>>>>>>> > interface ShardingFunction[UserT, DestinationT, ShardKeyT] >>>>>>>> extends >>>>>>>> > Serializable { >>>>>>>> > ShardKeyT assign(DestinationT destination, UserT element, >>>>>>>> > shardCount: Integer); >>>>>>>> > } >>>>>>>> > >>>>>>>> > Default implementation can be what is right now => random >>>>>>>> shard >>>>>>>> > encapsulated as ShardedKey<Integer>. >>>>>>>> > >>>>>>>> > On Thu, Apr 25, 2019 at 4:07 PM Reuven Lax <re...@google.com >>>>>>>> > <mailto:re...@google.com>> wrote: >>>>>>>> > >>>>>>>> > If sharding is not specified, then the semantics are >>>>>>>> > "runner-determined sharding." The DataflowRunner already >>>>>>>> takes >>>>>>>> > advantage of this to impose its own sharding if the user >>>>>>>> hasn't >>>>>>>> > specified an explicit one. Could the Flink runner do the >>>>>>>> same >>>>>>>> > instead of pushing this to the users? >>>>>>>> > >>>>>>>> > On Thu, Apr 25, 2019 at 6:23 AM Maximilian Michels >>>>>>>> > <m...@apache.org <mailto:m...@apache.org>> wrote: >>>>>>>> > >>>>>>>> > Hi Jozef, >>>>>>>> > >>>>>>>> > For sharding in FileIO there are basically two >>>>>>>> options: >>>>>>>> > >>>>>>>> > (1) num_shards ~= num_workers => bad spread of the >>>>>>>> load >>>>>>>> > across workers >>>>>>>> > (2) num_shards >> num_workers => good spread of the >>>>>>>> load >>>>>>>> > across workers, >>>>>>>> > but huge number of files >>>>>>>> > >>>>>>>> > Your approach would give users control over the >>>>>>>> sharding >>>>>>>> > keys such that >>>>>>>> > they could be adjusted to spread load more evenly. >>>>>>>> > >>>>>>>> > I'd like to hear from Beam IO experts if that would >>>>>>>> make sense. >>>>>>>> > >>>>>>>> > Thanks, >>>>>>>> > Max >>>>>>>> > >>>>>>>> > On 25.04.19 08:52, Jozef Vilcek wrote: >>>>>>>> > > Hello, >>>>>>>> > > >>>>>>>> > > Right now, if someone needs sharded files via >>>>>>>> FileIO, >>>>>>>> > there is only one >>>>>>>> > > option which is random (round robin) shard >>>>>>>> assignment per >>>>>>>> > element and it >>>>>>>> > > always use ShardedKey<Integer> as a key for the >>>>>>>> GBK which >>>>>>>> > follows. >>>>>>>> > > >>>>>>>> > > I would like to generalize this and have a >>>>>>>> possibility to >>>>>>>> > provide some >>>>>>>> > > ShardingFn[UserT, DestinationT, ShardKeyT] via >>>>>>>> FileIO. >>>>>>>> > > What I am mainly after is, to have a possibility to >>>>>>>> > provide optimisation >>>>>>>> > > for Flink runtime and pass in a special function >>>>>>>> which >>>>>>>> > generates shard >>>>>>>> > > keys in a way that they are evenly spread among >>>>>>>> workers >>>>>>>> > (BEAM-5865). >>>>>>>> > > >>>>>>>> > > Would such extension for FileIO make sense? If >>>>>>>> yes, I >>>>>>>> > would create a >>>>>>>> > > ticket for it and try to draft a PR. >>>>>>>> > > >>>>>>>> > > Best, >>>>>>>> > > Jozef >>>>>>>> > >>>>>>>> >>>>>>>
