A bunch of people have commented on the doc, without it seems any major disagreement. The PR is out for review.
On Fri, Sep 29, 2017 at 1:53 PM Eugene Kirpichov <kirpic...@google.com> wrote: > Hi all, > > Please take a look at some notes from a discussion we had about this with > a few folks, and an updated proposal and a couple of demo PRs implementing > the proposal. > http://s.apache.org/context-fn > > I hope this proposal is more agreeable. > > On Wed, Sep 13, 2017 at 3:46 PM Kenneth Knowles <k...@google.com.invalid> > wrote: > >> ValueProvider is global, PCollectionView is per-window, state is >> per-step/key/window, etc. >> >> So my unhappiness increases as we move through that list, adding more and >> more constraints on correct use, none of which are reflected in the API. >> Your description of "its context is an execution of the pipeline" is >> accurate for ValueProvider. The question is not merely "which DoFn will >> need which side inputs" but in which methods the side input is accessed >> (forbidden in every DoFn method other than @ProcessElement and @OnTimer). >> >> As for lambdas being more universal - I agree! But the capabilities of >> ParDo are not. I don't think we should transparently make them available >> anywhere you have a lambda. For example, multiply triggered side inputs >> fundamentally alter the semantics of MapElements and Filter to vary over >> time. The only reason this isn't a showstopper is that multiply triggered >> side inputs have very loose consistency already, and you can write >> nondeterministic predicates and map functions anyhow. If either of those >> were better, we'd want to keep them that way. >> >> Since NewDoFn is somewhat tied to the alternative proposal, and there's >> the >> point that since lambdas are cross-language we might reconsider >> ProcessContext (aka "pile of mud") style. But this universality - being >> the >> lowest common denominator across languages - is not a goal. Python already >> is quite different from Java, using | and >> and kwarg side inputs to good >> effect. And those two languages are quite similar. Go will look entirely >> different. For Java, annotation-driven APIs are common and offer important >> advantages for readability, validation, and forward/backward >> compatibility. >> And incidentally NewDoFn subsumes ProcessContext. >> >> On Wed, Sep 13, 2017 at 2:32 PM, Eugene Kirpichov < >> kirpic...@google.com.invalid> wrote: >> >> > Thanks! >> > >> > I think most of the issues you point out [validation, scheduling, >> > prefetching] are in the area of wiring. I reiterate that they can be >> solved >> > - both of the methods below will give the runner an answer to the >> low-level >> > question "which DoFn will need which side inputs": >> > >> > 1) Providing withSideInputs() builder methods on transforms that are >> > parameterized by user code. If only some side inputs should be made >> > available to particular bits of user code, provide more detailed >> > withBlahSideInputs() methods - this is up to the transform. >> > >> > 2) Inferring this from something annotation-driven as indicated in the >> > thread, e.g. capturing the PCollectionView in @SideInput-annotated >> public >> > fields. This can't be done on a lambda, because lambdas don't have >> fields >> > [so I think method #1 will keep being necessary], but it can be done on >> an >> > anonymous class. >> > >> > As for direct access being misleading: I'm not sure I agree. I think the >> > intuition for PCollectionView.get() is no more wrong than the intuition >> for >> > ValueProvider.get(): the return value is, logically, context-free [more >> > like: its context is an execution of the pipeline], so I have no issue >> with >> > it being accessed implicitly. >> > >> > On Wed, Sep 13, 2017 at 2:05 PM Kenneth Knowles <k...@google.com.invalid >> > >> > wrote: >> > >> > > I made some comments on >> https://issues.apache.org/jira/browse/BEAM-2950 >> > > which was filed to do a similar thing for State. Luke correctly >> pointed >> > out >> > > that many of the points apply here as well. I said most of the same >> > above, >> > > but I thought I'd pull them out again from that ticket and rephrase to >> > > apply to side inputs: >> > > >> > > - Direct access at first appears "more intuitive" because to a >> newcomer >> > it >> > > "looks like" normal [captured variable] access. But in fact it is >> nothing >> > > like normal [captured variable] access so this intuition is misleading >> > and >> > > should not be encouraged. So it is actually less readable because your >> > > intuitive reading is wrong. >> > > >> > > - This design would miss the validation aspect. One way it is >> different >> > > than normal [functional] programming is that there are many places it >> is >> > > illegal to reference [side inputs], such as StartBundle/FinishBundle, >> or >> > > passing to another object. This proposal would turn those into dynamic >> > > failures at best, or in the worst case data corruption (runner fails >> to >> > > catch illegal access, and permits some thread-global context to leak) >> > > >> > > - It is actually mandatory that we are always able to detect [side >> > inputs, >> > > or the user has to manually wire them], as it [must be scheduled >> > > differently] >> > > >> > > - A runner can't automatically prefetch, because it doesn't know >> which >> > > [side input] is used by which methods. >> > > >> > > - Magic by mutating stuff into place is just less readable / more >> error >> > > prone. >> > > >> > > State has even more compelling issues and none of the benefits so my >> > +0.75 >> > > for side inputs (now I am feeling more like +0.25) is a -1 for state. >> We >> > > should definitely not block one feature on all vaguely similar >> features. >> > > >> > > Kenn >> > > >> > > >> > > >> > > On Wed, Sep 13, 2017 at 1:56 PM, Eugene Kirpichov < >> > > kirpic...@google.com.invalid> wrote: >> > > >> > > > On Wed, Sep 13, 2017 at 1:44 PM Robert Bradshaw >> > > > <rober...@google.com.invalid> >> > > > wrote: >> > > > >> > > > > On Wed, Sep 13, 2017 at 1:17 PM, Eugene Kirpichov >> > > > > <kirpic...@google.com.invalid> wrote: >> > > > > > Hi Robert, >> > > > > > >> > > > > > Given the anticipated usage of this proposal in Java, I'm not >> sure >> > > the >> > > > > > Python approach you quoted is the right one. >> > > > > >> > > > > Perhaps not, but does that mean it would be a Java-ism only or >> would >> > > > > we implement it in Python despite it being worse there? >> > > > > >> > > > I'm not sure, but I don't see why the proposed approach of >> view.get() >> > > > wouldn't work well, or be harder to implement in Python. >> > > > >> > > > >> > > > > >> > > > > > The main reason: I see how it works with Map/FlatMap, but what >> > about >> > > > > cases >> > > > > > like FileIO.write(), parameterized by several lambdas (element >> -> >> > > > > > destination, destination -> filename policy, destination -> >> sink), >> > > > where >> > > > > > different lambdas may want to access different side inputs? It >> > feels >> > > > > > excessive to make each of the lambdas take all of the side >> inputs >> > in >> > > > the >> > > > > > same order; moreover, if the composite transform internally >> needs >> > to >> > > > pass >> > > > > > some more side inputs to the DoFn's executing these lambdas, it >> > will >> > > > need >> > > > > > to manipulate the argument lists in nontrivial ways to make >> sure it >> > > > > passes >> > > > > > them only the side inputs the user asked for, and in the proper >> > > order. >> > > > > >> > > > > In Python it would be trivial to "slice" the side input arguments >> > > > > across the lambdas in a natural way, but I can see that this >> would be >> > > > > more of a pain in Java, especially as lambdas are unnecessarily >> > > > > crippled during compilation. >> > > > > >> > > > > > Another reason is, I think with Java's type system it's >> impossible >> > to >> > > > > have >> > > > > > a NewDoFn-style API for lambdas, because annotations on lambda >> > > > arguments >> > > > > > are dropped when the lambda is converted to the respective >> > > > single-method >> > > > > > interface - a lambda is subject to a lot more type erasure than >> > > > anonymous >> > > > > > class. >> > > > > >> > > > > Yeah, this is unfortunate. But, as mentioned, side inputs don't >> need >> > > > > to be annotated, just counted. For something like inspecting the >> > > > > window the NewDoFn has a lot of advantages over implicit access >> (and >> > > > > makes it so you can't "forget" to declare your dependency), but I >> do >> > > > > see advantages for the implicit way of doing things for >> delegating to >> > > > > other callables. >> > > > > >> > > > > On the other hand, there is a bit of precedence for this: metrics >> > have >> > > > > the "implicit" api. If we do go this direction for side inputs, we >> > > > > should also consider it for state and side outputs. >> > > > > >> > > > I think Kenn is very strongly against using it for state, whereas I >> > don't >> > > > have an opinion either way because I can't think of a use case for >> > > > accessing state from a lambda - we should probably discuss this >> > > separately, >> > > > with proposed code examples in front of us. >> > > > >> > > > For side outputs, yes, it might be nice to ".add()" to a >> PCollection, >> > but >> > > > it would require bigger changes - e.g. creating intermediate >> > > PCollection's >> > > > and inserting an implicit Flatten in front of all steps that >> contribute >> > > to >> > > > this PCollection, because a PCollection currently can be produced >> only >> > > by 1 >> > > > step. Maybe there's a different way to express implicit side >> outputs. >> > > > Either way I support the idea of looking for such a way because it >> > would >> > > > simplify use cases such as error handling dead-letter collections. >> > > > >> > > > I guess the bigger point is: do we want to block the discussion of >> > > implicit >> > > > side inputs on making a decision about the implicitness of other >> things >> > > > (side outputs, state, PipelineOptions, window etc). I can see the >> > > argument >> > > > for a "yes, block", but can also see the argument for a "no, don't >> > > block" - >> > > > because this proposal is (as indicated earlier in the thread) >> > > > forward-compatible with annotation-based wiring, because we already >> > have >> > > a >> > > > precedent for implicit access of something via ValueProvider, and >> > because >> > > > of the advantages it offers. >> > > > >> > > > Want to mention another advantage: lambdas are likely to be much >> easier >> > > > than NewDoFn approach to use from non-Java but JVM languages/SDKs >> (e.g. >> > > > Scio), which might have even more type erasure, or might have less >> > > > sophisticated annotation machinery, or NewDoFn-style anonymous >> classes >> > > > might be highly non-idiomatic in them. Lambdas are idiomatic in >> every >> > > > language that supports lambdas, which these days is basically every >> > > > language. [I might be opening a can of worms here, but I guess you >> can >> > > > consider this an argument against NewDoFn in general - though that's >> > > > certainly outside the scope of this thread]. >> > > > >> > > > >> > > > > >> > > > > > On Wed, Sep 13, 2017 at 1:03 PM Robert Bradshaw >> > > > > <rober...@google.com.invalid> >> > > > > > wrote: >> > > > > > >> > > > > >> +1 to reducing the amount of boilerplate for dealing with side >> > > inputs. >> > > > > >> >> > > > > >> I prefer the "NewDoFn" style of side inputs for consistency. >> The >> > > > > >> primary drawback seems to be lambda's incompatibility with >> > > > > >> annotations. This is solved in Python by letting all the first >> > > > > >> annotated argument of the process method be the main input, and >> > > > > >> subsequent ones be the side input. For example >> > > > > >> >> > > > > >> main_pcoll | beam.Map( >> > > > > >> lambda main_input_elem, side_input_value: main_input_elem + >> > > > > >> side_input_value, >> > > > > >> side_input_pvalue) >> > > > > >> >> > > > > >> For multiple side inputs they are mapped positionally (though >> > Python >> > > > > >> has the advantage that arguments can be passed by keyword as >> well >> > to >> > > > > >> enhance readability when there are many of them, and we allow >> that >> > > for >> > > > > >> side inputs). Note that side_input_pvalue is not referenced >> > anywhere >> > > > > >> else, so we don't even have to store it and pass it around (one >> > > > > >> typically writes pvalue.AsList(some_pcoll) inline here). When >> the >> > > > > >> concrete PCollectionView is used to access the value this means >> > that >> > > > > >> it must be passed separately to both the ParDo and the callback >> > > > > >> (unless we can infer it, which I don't think we can do in all >> > > (many?) >> > > > > >> cases). >> > > > > >> >> > > > > >> There's no reason we couldn't do this, or something very >> similar, >> > in >> > > > > >> Java as well. >> > > > > >> >> > > > > >> On Wed, Sep 13, 2017 at 10:55 AM, Reuven Lax >> > > <re...@google.com.invalid >> > > > > >> > > > > >> wrote: >> > > > > >> > On Wed, Sep 13, 2017 at 10:05 AM, Eugene Kirpichov < >> > > > > >> > kirpic...@google.com.invalid> wrote: >> > > > > >> > >> > > > > >> >> Hi, >> > > > > >> >> >> > > > > >> >> I agree with these concerns to an extent, however I think >> the >> > > > > advantage >> > > > > >> of >> > > > > >> >> transparently letting any user code access side inputs, >> > > especially >> > > > > >> >> including lambdas, is so great that we should find a way to >> > > address >> > > > > >> these >> > > > > >> >> concerns within the constraints of the pattern I'm >> proposing. >> > See >> > > > > more >> > > > > >> >> below. >> > > > > >> >> >> > > > > >> >> On Wed, Sep 13, 2017 at 9:29 AM Ben Chambers >> > > > > >> <bchamb...@google.com.invalid >> > > > > >> >> > >> > > > > >> >> wrote: >> > > > > >> >> >> > > > > >> >> > One possible issue with this is that updating a thread >> local >> > is >> > > > > >> likely to >> > > > > >> >> > be much more expensive than passing an additional >> argument. >> > > > > >> >> >> > > > > >> >> This is an implementation detail that can be fixed - Luke >> made >> > a >> > > > > >> suggestion >> > > > > >> >> on the PR to set up the side input context once per bundle >> > rather >> > > > > than >> > > > > >> once >> > > > > >> >> per element. >> > > > > >> >> >> > > > > >> > >> > > > > >> > However remember that bundles might be small. Dataflow >> streaming >> > > > > runner >> > > > > >> > creates small bundles by design. The Flink runner creates >> > > > > single-element >> > > > > >> > bundles. >> > > > > >> > >> > > > > >> > >> > > > > >> >> >> > > > > >> >> >> > > > > >> >> > Also, not all >> > > > > >> >> > code called from within the DoFn will necessarily be in >> the >> > > same >> > > > > >> thread >> > > > > >> >> > (eg., sometimes we create a pool of threads for doing >> work). >> > > > > >> >> >> > > > > >> >> I think we already require that c.output() can not be done >> from >> > > > > multiple >> > > > > >> >> threads; and I don't think we document c.sideInput() to be >> > > > > thread-safe >> > > > > >> - it >> > > > > >> >> may be reasonable to declare that it isn't and has to be >> > accessed >> > > > > from >> > > > > >> the >> > > > > >> >> same thread as the ProcessElement call. If we want to relax >> > this, >> > > > > then >> > > > > >> >> there might be ways to deal with that too, e.g. provide >> > utilities >> > > > for >> > > > > >> the >> > > > > >> >> user to capture the "user code context" and restoring it >> > inside a >> > > > > >> thread. >> > > > > >> >> This would likely be valuable for other purposes, such as >> > making >> > > > > those >> > > > > >> >> extra threads visible to our profiling utilities. >> > > > > >> >> >> > > > > >> > >> > > > > >> > This seems fair, but we should be be very careful about our >> > > > > >> documentation. >> > > > > >> > And +1 to adding utilities to make multi-threaded work >> easier to >> > > > > manage. >> > > > > >> > >> > > > > >> >> >> > > > > >> >> >> > > > > >> >> > It may be >> > > > > >> >> > *more* confusing for this to sometimes work magically and >> > > > sometimes >> > > > > >> fail >> > > > > >> >> > horribly. Also, requiring the PCollectionView to be >> passed to >> > > > user >> > > > > >> code >> > > > > >> >> > that accesses it is nice because it makes *very clear* >> that >> > the >> > > > > side >> > > > > >> >> input >> > > > > >> >> > needs to be provided from the DoFn to that particular >> > utility. >> > > If >> > > > > it >> > > > > >> is >> > > > > >> >> > accessed via "spooky action at a distance" we lose that >> piece >> > > of >> > > > > >> "free" >> > > > > >> >> > documentation, which may lead to extensive misuse of these >> > > > utility >> > > > > >> >> methods. >> > > > > >> >> > >> > > > > >> >> I'd like to understand this concern better - from this >> > > description >> > > > > it's >> > > > > >> not >> > > > > >> >> clear to me. The pattern I'm proposing is that, when you're >> > > > > authoring a >> > > > > >> >> PTransform that is configured by any user callbacks, then: >> > > > > >> >> - you should provide a builder method .withSideInputs(...) >> > > > > >> >> - you should propagate those side inputs to all your >> internal >> > > > DoFn's >> > > > > >> that >> > > > > >> >> invoke the user code >> > > > > >> >> - in return the user callbacks will be allowed to access >> those >> > > > > >> particular >> > > > > >> >> side inputs >> > > > > >> >> This seems like a simple enough model to me to understand, >> both >> > > > from >> > > > > a >> > > > > >> >> user's perspective and from a transform author's >> perspective. >> > > > Steps 1 >> > > > > >> and 2 >> > > > > >> >> may eventually be automated by annotation analysis or other >> > means >> > > > > (e.g. >> > > > > >> SDK >> > > > > >> >> giving a way to provide given side inputs automatically to >> > > > everything >> > > > > >> >> inside a composite transform rather than to individual >> DoFn's). >> > > > > >> >> >> > > > > >> >> >> > > > > >> >> > >> > > > > >> >> > On Wed, Sep 6, 2017 at 11:10 AM Eugene Kirpichov >> > > > > >> >> > <kirpic...@google.com.invalid> wrote: >> > > > > >> >> > >> > > > > >> >> > > Hi, >> > > > > >> >> > > >> > > > > >> >> > > On Wed, Sep 6, 2017 at 10:55 AM Kenneth Knowles >> > > > > >> <k...@google.com.invalid >> > > > > >> >> > >> > > > > >> >> > > wrote: >> > > > > >> >> > > >> > > > > >> >> > > > On Wed, Sep 6, 2017 at 8:15 AM, Eugene Kirpichov < >> > > > > >> >> > > > kirpic...@google.com.invalid> wrote: >> > > > > >> >> > > > >> > > > > >> >> > > > > >> > > > > >> >> > > > > The differences are: >> > > > > >> >> > > > > - The proposal in the doc allows wiring different >> side >> > > > > inputs to >> > > > > >> >> the >> > > > > >> >> > > same >> > > > > >> >> > > > > Supplier, but I'm not convinced that this is >> important >> > - >> > > > you >> > > > > can >> > > > > >> >> just >> > > > > >> >> > > as >> > > > > >> >> > > > > easily call the constructor of your DoFn passing >> > > different >> > > > > >> >> > > > > PCollectionView's for it to capture. >> > > > > >> >> > > > > >> > > > > >> >> > > > >> > > > > >> >> > > > I disagree with this bit about it being "just as >> easy". >> > > > Passing >> > > > > >> the >> > > > > >> >> > > needed >> > > > > >> >> > > > PCollectionViews to your constructor (or even having a >> > > > > >> constructor) >> > > > > >> >> is >> > > > > >> >> > a >> > > > > >> >> > > > pain. Every time I have to do it, it adds a ton of >> > > > boilerplate >> > > > > >> that >> > > > > >> >> > feels >> > > > > >> >> > > > like pure noise. To make a DoFn reusable it must be >> made >> > > > into a >> > > > > >> named >> > > > > >> >> > > class >> > > > > >> >> > > > with a constructor, versus inlined with no >> constructor. >> > > > > >> >> > > >> > > > > >> >> > > Hm, why? You can have the DoFn be an anonymous class >> > > capturing >> > > > > the >> > > > > >> >> > > PCollectionView into a @SideInput field as a closure. >> > > > > >> >> > > >> > > > > >> >> > > >> > > > > >> >> > > > A generous analogy >> > > > > >> >> > > > is is that it is "just" manual closure >> > conversion/currying, >> > > > > >> changing >> > > > > >> >> > > > f(side, main) to f(side)(main). But in practice in >> Beam >> > the >> > > > > second >> > > > > >> >> one >> > > > > >> >> > > has >> > > > > >> >> > > > much more boilerplate. >> > > > > >> >> > > > >> > > > > >> >> > > > Also, Beam is worse. We present the user with >> > higher-order >> > > > > >> functions, >> > > > > >> >> > > which >> > > > > >> >> > > > is where the actual annoyance comes in. When you want >> to >> > > > > pardo(f) >> > > > > >> you >> > > > > >> >> > > have >> > > > > >> >> > > > to write pardo(f(side))(side, main). Your proposal is >> to >> > > > > support >> > > > > >> >> > > > pardo(f(side))(main) and mine is to support >> > pardo(f)(side, >> > > > > main). >> > > > > >> I >> > > > > >> >> > still >> > > > > >> >> > > > propose that we support both (as they get >> implemented). >> > If >> > > > you >> > > > > >> buy in >> > > > > >> >> > to >> > > > > >> >> > > my >> > > > > >> >> > > > analogy, then there's decades of precedent and the >> burden >> > > of >> > > > > proof >> > > > > >> >> > falls >> > > > > >> >> > > > heavily on whoever doesn't want to support both. >> > > > > >> >> > > > >> > > > > >> >> > > I see your point. I think the proposal is compatible >> with >> > > what >> > > > > >> you're >> > > > > >> >> > > suggesting too - in DoFn we could have @SideInput >> > > *parameters* >> > > > of >> > > > > >> type >> > > > > >> >> > > PCollectionView, with the same semantics as a field. >> > > > > >> >> > > >> > > > > >> >> > > >> > > > > >> >> > > > >> > > > > >> >> > > > - My proposal allows getting rid of .withSideInputs() >> > > > entirely, >> > > > > >> >> because >> > > > > >> >> > > the >> > > > > >> >> > > > > DoFn captures the PCollectionView so you don't need >> to >> > > > > specify >> > > > > >> it >> > > > > >> >> > > > > explicitly for wiring. >> > > > > >> >> > > > > >> > > > > >> >> > > > >> > > > > >> >> > > > I've decided to change to full +1 (whatever that means >> > > > > compared to >> > > > > >> >> 0.75 >> > > > > >> >> > > :-) >> > > > > >> >> > > > to adding support for @SideInput fields, because the >> > > benefits >> > > > > >> >> outweigh >> > > > > >> >> > > this >> > > > > >> >> > > > failure mode: >> > > > > >> >> > > > >> > > > > >> >> > > > new DoFn { >> > > > > >> >> > > > // forgot the annotation >> > > > > >> >> > > > private final PCollectionView whatever; >> > > > > >> >> > > > >> > > > > >> >> > > > @ProcessElement public void process(...) { >> > > > > >> >> > > > whatever.get(); // crash during execution >> > > > > >> >> > > > } >> > > > > >> >> > > > } >> > > > > >> >> > > > >> > > > > >> >> > > > But ideas to mitigate that would be cool. >> > > > > >> >> > > >> > > > > >> >> > > Hm, can't think of anything less hacky than "prohibit >> > having >> > > > > fields >> > > > > >> of >> > > > > >> >> > type >> > > > > >> >> > > PCollectionView that are not public, final, and >> annotated >> > > with >> > > > > >> >> > @SideInput" >> > > > > >> >> > > - not sure we'd want to go down this road. I suppose a >> good >> > > > error >> > > > > >> >> message >> > > > > >> >> > > in .get() would be sufficient, saying "Did you forget to >> > > > specify >> > > > > a >> > > > > >> >> > > requirement for this side input via .withSideInputs() >> or by >> > > > > >> annotating >> > > > > >> >> > the >> > > > > >> >> > > field as @SideInput" or something like that. >> > > > > >> >> > > >> > > > > >> >> > > > >> > > > > >> >> > > >> > > > > >> >> > > >> > > > > >> >> > > > Kenn >> > > > > >> >> > > > >> > > > > >> >> > > > >> > > > > >> >> > > > > >> > > > > >> >> > > > > On Wed, Sep 6, 2017 at 6:03 AM Lukasz Cwik >> > > > > >> >> <lc...@google.com.invalid >> > > > > >> >> > > >> > > > > >> >> > > > > wrote: >> > > > > >> >> > > > > >> > > > > >> >> > > > > > My concern with the proposal is not the specifics >> of >> > > how >> > > > it >> > > > > >> will >> > > > > >> >> > work >> > > > > >> >> > > > and >> > > > > >> >> > > > > > more about it being yet another way on how our >> API is >> > > to >> > > > be >> > > > > >> used >> > > > > >> >> > even >> > > > > >> >> > > > > > though we have a proposal [1] of an API style we >> were >> > > > > working >> > > > > >> >> > towards >> > > > > >> >> > > > in >> > > > > >> >> > > > > > Java and Python. I would rather re-open that >> > discussion >> > > > now >> > > > > >> about >> > > > > >> >> > > what >> > > > > >> >> > > > we >> > > > > >> >> > > > > > want that API to look like for our major features >> and >> > > > work >> > > > > >> >> towards >> > > > > >> >> > > > > > consistency (or not if there is a strong argument >> as >> > to >> > > > why >> > > > > >> some >> > > > > >> >> > > > feature >> > > > > >> >> > > > > > should have a different style). >> > > > > >> >> > > > > > >> > > > > >> >> > > > > > 1: https://s.apache.org/a-new-dofn >> > > > > >> >> > > > > > >> > > > > >> >> > > > > > On Wed, Sep 6, 2017 at 12:22 AM, Kenneth Knowles >> > > > > >> >> > > > <k...@google.com.invalid >> > > > > >> >> > > > > > >> > > > > >> >> > > > > > wrote: >> > > > > >> >> > > > > > >> > > > > >> >> > > > > > > +0.75 because I'd like to bring up invalid >> > pipelines. >> > > > > >> >> > > > > > > >> > > > > >> >> > > > > > > I had proposed side inputs as parameters to >> DoFn in >> > > > > >> >> > > > > > > https://s.apache.org/a-new-dofn (specifically >> at >> > > [1]) >> > > > so >> > > > > >> the >> > > > > >> >> > only >> > > > > >> >> > > > > place >> > > > > >> >> > > > > > > they are specified is in the graph construction, >> > > making >> > > > > the >> > > > > >> >> DoFn >> > > > > >> >> > > more >> > > > > >> >> > > > > > > reusable and errors impossible. I've actually >> been >> > > > > noodling >> > > > > >> my >> > > > > >> >> > way >> > > > > >> >> > > > > > towards >> > > > > >> >> > > > > > > this in a branch :-) >> > > > > >> >> > > > > > > >> > > > > >> >> > > > > > > Eugene's proposal is a sort of converse, where >> the >> > > side >> > > > > >> inputs >> > > > > >> >> > are >> > > > > >> >> > > > > values >> > > > > >> >> > > > > > > captured in the closure and not parameters, yet >> the >> > > > only >> > > > > >> place >> > > > > >> >> > they >> > > > > >> >> > > > are >> > > > > >> >> > > > > > > specified is in the DoFn. >> > > > > >> >> > > > > > > >> > > > > >> >> > > > > > > I see no conflict between these two. It is very >> > > natural >> > > > > to >> > > > > >> have >> > > > > >> >> > > both >> > > > > >> >> > > > > the >> > > > > >> >> > > > > > > capability to accept parameters and the ability >> to >> > > > > capture >> > > > > >> >> > > variables >> > > > > >> >> > > > in >> > > > > >> >> > > > > > the >> > > > > >> >> > > > > > > closure. Supporting both is totally standard in >> > > > > up-to-date >> > > > > >> >> > > > programming >> > > > > >> >> > > > > > > languages. >> > > > > >> >> > > > > > > >> > > > > >> >> > > > > > > Today we have the worse of both worlds: >> > > PCollectionView >> > > > > >> behaves >> > > > > >> >> > as >> > > > > >> >> > > > > > > something captured in the closure/constructor, >> but >> > > must >> > > > > >> still >> > > > > >> >> be >> > > > > >> >> > > > > > explicitly >> > > > > >> >> > > > > > > wired up. >> > > > > >> >> > > > > > > >> > > > > >> >> > > > > > > But if I use PCollectionView.get() and have not >> > wired >> > > > it >> > > > > up >> > > > > >> in >> > > > > >> >> > any >> > > > > >> >> > > > way, >> > > > > >> >> > > > > > > what happens? Just like today, you can try to >> > > > > >> .sideInput(...) a >> > > > > >> >> > > thing >> > > > > >> >> > > > > > that >> > > > > >> >> > > > > > > is not available. With side inputs as >> parameters, >> > > this >> > > > is >> > > > > >> not >> > > > > >> >> > > > possible. >> > > > > >> >> > > > > > If >> > > > > >> >> > > > > > > you want to treat them as captured in a closure, >> > > while >> > > > > >> avoiding >> > > > > >> >> > > > errors, >> > > > > >> >> > > > > > it >> > > > > >> >> > > > > > > seems like you might need to do some low-level >> > magic, >> > > > > like >> > > > > >> the >> > > > > >> >> > > > > > > serialization-based detection that Luke has >> > suggested >> > > > > before >> > > > > >> >> > (there >> > > > > >> >> > > > are >> > > > > >> >> > > > > > > known downsides that we haven't explored, like >> > > > > overcapture). >> > > > > >> >> > > > > > > >> > > > > >> >> > > > > > > Kenn >> > > > > >> >> > > > > > > >> > > > > >> >> > > > > > > [1] >> > > > > >> >> > > > > > > >> > > > > >> >> > https://docs.google.com/document/d/ >> > > > 1ClmQ6LqdnfseRzeSw3SL68DAO1f8j >> > > > > >> >> > > > > > > sWBL2FfzWErlbw/edit#heading=h.1budnm7l01ko >> > > > > >> >> > > > > > > >> > > > > >> >> > > > > > > >> > > > > >> >> > > > > > > On Tue, Sep 5, 2017 at 11:24 PM, Eugene >> Kirpichov < >> > > > > >> >> > > > > > > kirpic...@google.com.invalid> wrote: >> > > > > >> >> > > > > > > >> > > > > >> >> > > > > > > > Hm, I guess you're right - for outputs it >> could >> > be >> > > > > indeed >> > > > > >> >> quite >> > > > > >> >> > > > > > valuable >> > > > > >> >> > > > > > > to >> > > > > >> >> > > > > > > > output to them without plumbing (e.g. >> outputting >> > > > > errors). >> > > > > >> >> Could >> > > > > >> >> > > be >> > > > > >> >> > > > > done >> > > > > >> >> > > > > > > > perhaps via TupleTag.output()? (assuming the >> same >> > > > > TupleTag >> > > > > >> >> can >> > > > > >> >> > > not >> > > > > >> >> > > > be >> > > > > >> >> > > > > > > > reused to tag multiple PCollection's) >> > > > > >> >> > > > > > > > >> > > > > >> >> > > > > > > > For now I sent a PR for side input support >> > > > > >> >> > > > > > > > https://github.com/apache/beam/pull/3814 . >> > > > > >> >> > > > > > > > >> > > > > >> >> > > > > > > > On Tue, Sep 5, 2017 at 9:52 PM Lukasz Cwik >> > > > > >> >> > > > <lc...@google.com.invalid >> > > > > >> >> > > > > > >> > > > > >> >> > > > > > > > wrote: >> > > > > >> >> > > > > > > > >> > > > > >> >> > > > > > > > > I disagree, state may not care where it is >> used >> > > as >> > > > > well >> > > > > >> >> > since a >> > > > > >> >> > > > > > person >> > > > > >> >> > > > > > > > may >> > > > > >> >> > > > > > > > > call a function which needs to >> store/retrieve >> > > state >> > > > > and >> > > > > >> >> > instead >> > > > > >> >> > > > of >> > > > > >> >> > > > > > > having >> > > > > >> >> > > > > > > > > the DoFn declare the StateSpec and then >> pass in >> > > the >> > > > > >> state >> > > > > >> >> > > > > > > implementation >> > > > > >> >> > > > > > > > > down into the function everywhere. Similarly >> > for >> > > > > >> outputs, >> > > > > >> >> the >> > > > > >> >> > > > > > internal >> > > > > >> >> > > > > > > > > functions could take the TupleTag and >> request >> > an >> > > > > output >> > > > > >> >> > manager >> > > > > >> >> > > > or >> > > > > >> >> > > > > > take >> > > > > >> >> > > > > > > > an >> > > > > >> >> > > > > > > > > "output" reference which give functions the >> > > ability >> > > > > to >> > > > > >> >> > produce >> > > > > >> >> > > > > output >> > > > > >> >> > > > > > > > > directly without needing to pass everything >> > that >> > > is >> > > > > >> needed >> > > > > >> >> to >> > > > > >> >> > > be >> > > > > >> >> > > > > > output >> > > > > >> >> > > > > > > > > back to the caller. >> > > > > >> >> > > > > > > > > >> > > > > >> >> > > > > > > > > On Tue, Sep 5, 2017 at 9:23 PM, Eugene >> > Kirpichov >> > > < >> > > > > >> >> > > > > > > > > kirpic...@google.com.invalid> wrote: >> > > > > >> >> > > > > > > > > >> > > > > >> >> > > > > > > > > > Hm, I think of these things (state, side >> > > outputs >> > > > > >> etc.), >> > > > > >> >> > only >> > > > > >> >> > > > side >> > > > > >> >> > > > > > > > inputs >> > > > > >> >> > > > > > > > > > make sense to access in arbitrary user >> > > callbacks >> > > > > >> without >> > > > > >> >> > > > explicit >> > > > > >> >> > > > > > > > > knowledge >> > > > > >> >> > > > > > > > > > of the surrounding transform - so only >> side >> > > > inputs >> > > > > >> can be >> > > > > >> >> > > > > implicit >> > > > > >> >> > > > > > > like >> > > > > >> >> > > > > > > > > > this. >> > > > > >> >> > > > > > > > > > >> > > > > >> >> > > > > > > > > > Ultimately we'll probably end up removing >> > > > > >> ProcessContext, >> > > > > >> >> > and >> > > > > >> >> > > > > > keeping >> > > > > >> >> > > > > > > > > only >> > > > > >> >> > > > > > > > > > annotations (on fields / methods / >> > parameters). >> > > > In >> > > > > >> that >> > > > > >> >> > > world, >> > > > > >> >> > > > a >> > > > > >> >> > > > > > > field >> > > > > >> >> > > > > > > > > > annotation could be used (like per my >> > previous >> > > > > email) >> > > > > >> to >> > > > > >> >> > > > > statically >> > > > > >> >> > > > > > > > > specify >> > > > > >> >> > > > > > > > > > which side inputs will be needed - while >> the >> > > > value >> > > > > >> could >> > > > > >> >> > > still >> > > > > >> >> > > > be >> > > > > >> >> > > > > > > > > accessed >> > > > > >> >> > > > > > > > > > via .get(), just like state cells are >> > accessed >> > > > via >> > > > > >> >> .read() >> > > > > >> >> > > and >> > > > > >> >> > > > > > > > .write(): >> > > > > >> >> > > > > > > > > > i.e., #get() is not a new method of >> access. >> > > > > >> >> > > > > > > > > > >> > > > > >> >> > > > > > > > > > Overall, it seems like I should proceed >> with >> > > the >> > > > > >> idea. I >> > > > > >> >> > > filed >> > > > > >> >> > > > > > > > > > >> > > https://issues.apache.org/jira/browse/BEAM-2844. >> > > > > >> >> > > > > > > > > > >> > > > > >> >> > > > > > > > > > On Tue, Sep 5, 2017 at 9:08 PM Lukasz Cwik >> > > > > >> >> > > > > > <lc...@google.com.invalid >> > > > > >> >> > > > > > > > >> > > > > >> >> > > > > > > > > > wrote: >> > > > > >> >> > > > > > > > > > >> > > > > >> >> > > > > > > > > > > For API consistency reasons, it would be >> > good >> > > > if >> > > > > we >> > > > > >> did >> > > > > >> >> > > this >> > > > > >> >> > > > > > > > > holistically >> > > > > >> >> > > > > > > > > > > and expanded this approach to state, >> side >> > > > > outputs, >> > > > > >> ... >> > > > > >> >> so >> > > > > >> >> > > > that >> > > > > >> >> > > > > a >> > > > > >> >> > > > > > > > person >> > > > > >> >> > > > > > > > > > can >> > > > > >> >> > > > > > > > > > > always call Something.get() to return >> > > something >> > > > > that >> > > > > >> >> they >> > > > > >> >> > > can >> > > > > >> >> > > > > > > access >> > > > > >> >> > > > > > > > > > > implementation wise. It will be >> confusing >> > for >> > > > our >> > > > > >> users >> > > > > >> >> > to >> > > > > >> >> > > > have >> > > > > >> >> > > > > > > many >> > > > > >> >> > > > > > > > > > > variations in our style of how all these >> > > > concepts >> > > > > >> are >> > > > > >> >> > used >> > > > > >> >> > > > > > > > > > (ProcessContext >> > > > > >> >> > > > > > > > > > > / Annotations / #get()) >> > > > > >> >> > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > On Tue, Sep 5, 2017 at 8:08 AM, Eugene >> > > > Kirpichov >> > > > > < >> > > > > >> >> > > > > > > > > > > kirpic...@google.com.invalid> wrote: >> > > > > >> >> > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > > Also, I think my approach is >> compatible >> > > with >> > > > > >> >> > annotations >> > > > > >> >> > > > and >> > > > > >> >> > > > > > > future >> > > > > >> >> > > > > > > > > > > removal >> > > > > >> >> > > > > > > > > > > > of .withSideInputs if we annotate a >> > field: >> > > > > >> >> > > > > > > > > > > > final PCollectionView<Foo> foo = ...; >> > > > > >> >> > > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > > class MyDoFn { >> > > > > >> >> > > > > > > > > > > > @SideInput >> > > > > >> >> > > > > > > > > > > > PCollectionView<Foo> foo = foo; >> > > > > >> >> > > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > > ...foo.get()... >> > > > > >> >> > > > > > > > > > > > } >> > > > > >> >> > > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > > We can extract the accessed views from >> > the >> > > > DoFn >> > > > > >> >> > instance >> > > > > >> >> > > > > using >> > > > > >> >> > > > > > > > > > > reflection. >> > > > > >> >> > > > > > > > > > > > Still not compatible with lambdas, but >> > > > > compatible >> > > > > >> >> > > > > automatically >> > > > > >> >> > > > > > > > with >> > > > > >> >> > > > > > > > > > all >> > > > > >> >> > > > > > > > > > > > anonymous classes. >> > > > > >> >> > > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > > On Tue, Sep 5, 2017, 8:02 AM Eugene >> > > > Kirpichov < >> > > > > >> >> > > > > > > > kirpic...@google.com> >> > > > > >> >> > > > > > > > > > > > wrote: >> > > > > >> >> > > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > > > Hi Luke, >> > > > > >> >> > > > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > > > I know this (annotations) is the >> > pattern >> > > we >> > > > > were >> > > > > >> >> > > > > considering >> > > > > >> >> > > > > > > for >> > > > > >> >> > > > > > > > > side >> > > > > >> >> > > > > > > > > > > > > inputs, but I no longer think it is >> the >> > > > best >> > > > > >> way to >> > > > > >> >> > > > access >> > > > > >> >> > > > > > > them. >> > > > > >> >> > > > > > > > > > > > > Annotations help getting rid of the >> > > > > >> >> .withSideInputs() >> > > > > >> >> > > > call, >> > > > > >> >> > > > > > but >> > > > > >> >> > > > > > > > > this >> > > > > >> >> > > > > > > > > > is >> > > > > >> >> > > > > > > > > > > > > where their advantage ends. >> > > > > >> >> > > > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > > > The advantages of the proposed >> approach >> > > are >> > > > > >> that it >> > > > > >> >> > > > > > > automatically >> > > > > >> >> > > > > > > > > > works >> > > > > >> >> > > > > > > > > > > > > with all existing callback or lambda >> > > code. >> > > > No >> > > > > >> need >> > > > > >> >> to >> > > > > >> >> > > > > further >> > > > > >> >> > > > > > > > > develop >> > > > > >> >> > > > > > > > > > > the >> > > > > >> >> > > > > > > > > > > > > reflection machinery to support side >> > > input >> > > > > >> >> > annotations >> > > > > >> >> > > - >> > > > > >> >> > > > > and >> > > > > >> >> > > > > > > > > > especially >> > > > > >> >> > > > > > > > > > > > to >> > > > > >> >> > > > > > > > > > > > > support arbitrary user interfaces, >> no >> > > need >> > > > to >> > > > > >> >> change >> > > > > >> >> > > > > existing >> > > > > >> >> > > > > > > > > > > transforms, >> > > > > >> >> > > > > > > > > > > > > no need for transform authors to >> even >> > > know >> > > > > that >> > > > > >> the >> > > > > >> >> > > > > machinery >> > > > > >> >> > > > > > > > > exists >> > > > > >> >> > > > > > > > > > to >> > > > > >> >> > > > > > > > > > > > > make side inputs usable in their >> > > transforms >> > > > > >> (and no >> > > > > >> >> > > need >> > > > > >> >> > > > > for >> > > > > >> >> > > > > > > > > authors >> > > > > >> >> > > > > > > > > > to >> > > > > >> >> > > > > > > > > > > > > think about whether or not they >> should >> > > > > support >> > > > > >> side >> > > > > >> >> > > > > inputs). >> > > > > >> >> > > > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > > > Moreover, like Reuven says, >> annotations >> > > > don't >> > > > > >> work >> > > > > >> >> > with >> > > > > >> >> > > > > > lambdas >> > > > > >> >> > > > > > > > at >> > > > > >> >> > > > > > > > > > all: >> > > > > >> >> > > > > > > > > > > > > creating a lambda with a flexible >> set >> > of >> > > > > >> annotation >> > > > > >> >> > > > > arguments >> > > > > >> >> > > > > > > > > appears >> > > > > >> >> > > > > > > > > > > to >> > > > > >> >> > > > > > > > > > > > be >> > > > > >> >> > > > > > > > > > > > > currently impossible, and even >> > capturing >> > > > the >> > > > > >> >> > > annotations >> > > > > >> >> > > > on >> > > > > >> >> > > > > > > > > arguments >> > > > > >> >> > > > > > > > > > > of >> > > > > >> >> > > > > > > > > > > > a >> > > > > >> >> > > > > > > > > > > > > lambda is I believe also impossible >> > > because >> > > > > the >> > > > > >> >> Java >> > > > > >> >> > > > > compiler >> > > > > >> >> > > > > > > > drops >> > > > > >> >> > > > > > > > > > > them >> > > > > >> >> > > > > > > > > > > > in >> > > > > >> >> > > > > > > > > > > > > the generated class or method >> handle. >> > > > > >> >> > > > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > > > On Tue, Sep 5, 2017 at 6:57 AM >> Lukasz >> > > Cwik >> > > > > >> >> > > > > > > > > <lc...@google.com.invalid >> > > > > >> >> > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > > > wrote: >> > > > > >> >> > > > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > > >> I believe we should follow the >> pattern >> > > > that >> > > > > >> state >> > > > > >> >> > uses >> > > > > >> >> > > > and >> > > > > >> >> > > > > > > add a >> > > > > >> >> > > > > > > > > > type >> > > > > >> >> > > > > > > > > > > > >> annotation to link the side input >> > > > > definition to >> > > > > >> >> its >> > > > > >> >> > > > usage >> > > > > >> >> > > > > > > > > directly. >> > > > > >> >> > > > > > > > > > > This >> > > > > >> >> > > > > > > > > > > > >> would allow us to know that the >> side >> > > input >> > > > > was >> > > > > >> >> > > > definitely >> > > > > >> >> > > > > > > being >> > > > > >> >> > > > > > > > > > > accessed >> > > > > >> >> > > > > > > > > > > > >> and perform validation during graph >> > > > > >> construction >> > > > > >> >> for >> > > > > >> >> > > any >> > > > > >> >> > > > > > used >> > > > > >> >> > > > > > > > but >> > > > > >> >> > > > > > > > > > > > >> unspecified side inputs. >> > > > > >> >> > > > > > > > > > > > >> >> > > > > >> >> > > > > > > > > > > > >> Code snippet: >> > > > > >> >> > > > > > > > > > > > >> final PCollectionView<String> foo = >> > > > > >> >> > > > > > pipeline.apply("fooName", >> > > > > >> >> > > > > > > > > > > > >> Create.of("foo")).apply(View.< >> > > > > >> >> String>asSingleton()); >> > > > > >> >> > > > > > > > > > > > >> PCollection<String> output = >> pipeline >> > > > > >> >> > > > > > > > > > > > >> .apply(Create.of(1, 2, 3)) >> > > > > >> >> > > > > > > > > > > > >> .apply(MapElements.via( >> > > > > >> >> > > > > > > > > > > > >> new SimpleFunction<Integer, >> > > > > String>() { >> > > > > >> >> > > > > > > > > > > > >> @Override >> > > > > >> >> > > > > > > > > > > > >> public String >> apply(Integer >> > > > input, >> > > > > >> >> > > > > > > > @SideInput("fooName") >> > > > > >> >> > > > > > > > > > > > String >> > > > > >> >> > > > > > > > > > > > >> fooValue) { >> > > > > >> >> > > > > > > > > > > > >> return fooValue + " " + >> > > input; >> > > > > >> >> > > > > > > > > > > > >> } >> > > > > >> >> > > > > > > > > > > > >> }).withSideInputs(foo));* >> > > > > >> >> > > > > > > > > > > > >> >> > > > > >> >> > > > > > > > > > > > >> On Mon, Sep 4, 2017 at 6:22 PM, >> Eugene >> > > > > >> Kirpichov < >> > > > > >> >> > > > > > > > > > > > >> kirpic...@google.com.invalid> >> wrote: >> > > > > >> >> > > > > > > > > > > > >> >> > > > > >> >> > > > > > > > > > > > >> > Sure, here's how a modified >> > (passing) >> > > > > >> >> MapElements >> > > > > >> >> > > unit >> > > > > >> >> > > > > > test >> > > > > >> >> > > > > > > > > looks >> > > > > >> >> > > > > > > > > > > > like, >> > > > > >> >> > > > > > > > > > > > >> > with usage of side inputs: >> > > > > >> >> > > > > > > > > > > > >> > >> > > > > >> >> > > > > > > > > > > > >> > @Test >> > > > > >> >> > > > > > > > > > > > >> > @Category(NeedsRunner.class) >> > > > > >> >> > > > > > > > > > > > >> > public void >> > > > testMapBasicWithSideInput() >> > > > > >> throws >> > > > > >> >> > > > > > Exception { >> > > > > >> >> > > > > > > > > > > > >> > * final >> PCollectionView<String> >> > foo >> > > > =* >> > > > > >> >> > > > > > > > > > > > >> > * pipeline.apply("foo", >> > > > > >> >> > > > > > > > > > > > >> > >> > > > > >> >> > > Create.of("foo")).apply(View.<String>asSingleton());* >> > > > > >> >> > > > > > > > > > > > >> > PCollection<String> output = >> > > > pipeline >> > > > > >> >> > > > > > > > > > > > >> > .apply(Create.of(1, 2, >> 3)) >> > > > > >> >> > > > > > > > > > > > >> > .apply(MapElements.via( >> > > > > >> >> > > > > > > > > > > > >> > new >> > > SimpleFunction<Integer, >> > > > > >> >> String>() >> > > > > >> >> > { >> > > > > >> >> > > > > > > > > > > > >> > @Override >> > > > > >> >> > > > > > > > > > > > >> > public String >> > > > apply(Integer >> > > > > >> >> input) { >> > > > > >> >> > > > > > > > > > > > >> > return* foo.get() >> > *+ " >> > > > " + >> > > > > >> >> input; >> > > > > >> >> > > > > > > > > > > > >> > } >> > > > > >> >> > > > > > > > > > > > >> > }) >> > > > > >> >> > > > > > > > > > > > >> > *.withSideInputs(foo));* >> > > > > >> >> > > > > > > > > > > > >> > >> > > > > >> >> > > > > > > > > > > > >> > PAssert.that(output). >> > > > > >> >> containsInAnyOrder("foo >> > > > > >> >> > 1", >> > > > > >> >> > > > > "foo >> > > > > >> >> > > > > > > 2", >> > > > > >> >> > > > > > > > > > "foo >> > > > > >> >> > > > > > > > > > > > 3"); >> > > > > >> >> > > > > > > > > > > > >> > pipeline.run(); >> > > > > >> >> > > > > > > > > > > > >> > } >> > > > > >> >> > > > > > > > > > > > >> > >> > > > > >> >> > > > > > > > > > > > >> > >> > > > > >> >> > > > > > > > > > > > >> > On Mon, Sep 4, 2017 at 6:12 PM >> > Reuven >> > > > Lax >> > > > > >> >> > > > > > > > > > <re...@google.com.invalid >> > > > > >> >> > > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > > >> > wrote: >> > > > > >> >> > > > > > > > > > > > >> > >> > > > > >> >> > > > > > > > > > > > >> > > Can you provide a code snippet >> > > showing >> > > > > how >> > > > > >> >> this >> > > > > >> >> > > > would >> > > > > >> >> > > > > > > look? >> > > > > >> >> > > > > > > > > > > > >> > > >> > > > > >> >> > > > > > > > > > > > >> > > On Sun, Sep 3, 2017 at 6:49 PM, >> > > Eugene >> > > > > >> >> > Kirpichov < >> > > > > >> >> > > > > > > > > > > > >> > > kirpic...@google.com.invalid> >> > > wrote: >> > > > > >> >> > > > > > > > > > > > >> > > >> > > > > >> >> > > > > > > > > > > > >> > > > TL;DR Introduce method >> > > > > >> >> PCollectionView.get(), >> > > > > >> >> > > > > > > implemented >> > > > > >> >> > > > > > > > > as: >> > > > > >> >> > > > > > > > > > > get >> > > > > >> >> > > > > > > > > > > > >> > > > thread-local ProcessContext >> and >> > do >> > > > > >> >> > > > > c.sideInput(this). >> > > > > >> >> > > > > > > As a >> > > > > >> >> > > > > > > > > > > result, >> > > > > >> >> > > > > > > > > > > > >> any >> > > > > >> >> > > > > > > > > > > > >> > > user >> > > > > >> >> > > > > > > > > > > > >> > > > lambdas such as MapElements >> can >> > > use >> > > > > side >> > > > > >> >> > inputs. >> > > > > >> >> > > > > > > > > > > > >> > > > >> > > > > >> >> > > > > > > > > > > > >> > > > Quite a few transforms have >> > > > user-code >> > > > > >> >> > callbacks >> > > > > >> >> > > or >> > > > > >> >> > > > > > > > lambdas: >> > > > > >> >> > > > > > > > > > > ParDo >> > > > > >> >> > > > > > > > > > > > >> > (DoFn), >> > > > > >> >> > > > > > > > > > > > >> > > > Map/FlatMapElements, the >> > > > > >> DynamicDestinations >> > > > > >> >> > > > classes >> > > > > >> >> > > > > > in >> > > > > >> >> > > > > > > > > > various >> > > > > >> >> > > > > > > > > > > > IOs, >> > > > > >> >> > > > > > > > > > > > >> > > > combine fns, the PollFn >> callback >> > > in >> > > > > >> Watch, >> > > > > >> >> > etc. >> > > > > >> >> > > > > > > > > > > > >> > > > >> > > > > >> >> > > > > > > > > > > > >> > > > Of these, only DoFn and >> > CombineFn >> > > > have >> > > > > >> >> > built-in >> > > > > >> >> > > > > > support >> > > > > >> >> > > > > > > > for >> > > > > >> >> > > > > > > > > > side >> > > > > >> >> > > > > > > > > > > > >> > inputs; >> > > > > >> >> > > > > > > > > > > > >> > > > for DynamicDestinations it is >> > > > plumbed >> > > > > >> >> > > explicitly; >> > > > > >> >> > > > > > others >> > > > > >> >> > > > > > > > > don't >> > > > > >> >> > > > > > > > > > > > have >> > > > > >> >> > > > > > > > > > > > >> > > access >> > > > > >> >> > > > > > > > > > > > >> > > > (e.g. you can't access side >> > inputs >> > > > > from >> > > > > >> >> > > > > > > > Map/FlatMapElements >> > > > > >> >> > > > > > > > > > > > because >> > > > > >> >> > > > > > > > > > > > >> > they >> > > > > >> >> > > > > > > > > > > > >> > > > don't have a ProcessContext >> or >> > any >> > > > > >> context >> > > > > >> >> for >> > > > > >> >> > > > that >> > > > > >> >> > > > > > > > matter). >> > > > > >> >> > > > > > > > > > > > >> > > > >> > > > > >> >> > > > > > > > > > > > >> > > > I think it's important to >> solve >> > > > this, >> > > > > >> >> > especially >> > > > > >> >> > > > as >> > > > > >> >> > > > > > > Java 8 >> > > > > >> >> > > > > > > > > > > becomes >> > > > > >> >> > > > > > > > > > > > >> > > people's >> > > > > >> >> > > > > > > > > > > > >> > > > default choice: users will >> want >> > to >> > > > use >> > > > > >> side >> > > > > >> >> > > inputs >> > > > > >> >> > > > > in >> > > > > >> >> > > > > > > > > > > > >> > > Map/FlatMapElements. >> > > > > >> >> > > > > > > > > > > > >> > > > >> > > > > >> >> > > > > > > > > > > > >> > > > It also appears to be quite >> easy >> > > to >> > > > > >> >> implement: >> > > > > >> >> > > > > > > > > > > > >> > > > >> > > > > >> >> > > > > > > > > > > > >> > > > Runner part: >> > > > > >> >> > > > > > > > > > > > >> > > > - introduce a >> SideInputAccessor >> > > > > interface >> > > > > >> >> > > > > > > > > > > > >> > > > - make .get() on a >> > PCollectionView >> > > > > get it >> > > > > >> >> > from a >> > > > > >> >> > > > > > > > > thread-local >> > > > > >> >> > > > > > > > > > > > >> > > > SideInputAccessor >> > > > > >> >> > > > > > > > > > > > >> > > > - make runners set the >> > > thread-local >> > > > > >> >> > > > > SideInputAccessor >> > > > > >> >> > > > > > > any >> > > > > >> >> > > > > > > > > time >> > > > > >> >> > > > > > > > > > > the >> > > > > >> >> > > > > > > > > > > > >> > runner >> > > > > >> >> > > > > > > > > > > > >> > > > is evaluating something in a >> > > context >> > > > > >> where >> > > > > >> >> > side >> > > > > >> >> > > > > inputs >> > > > > >> >> > > > > > > are >> > > > > >> >> > > > > > > > > > > > >> available, >> > > > > >> >> > > > > > > > > > > > >> > > e.g. >> > > > > >> >> > > > > > > > > > > > >> > > > a ProcessElement method, or >> > > > applying a >> > > > > >> >> > > CombineFn - >> > > > > >> >> > > > > the >> > > > > >> >> > > > > > > set >> > > > > >> >> > > > > > > > > of >> > > > > >> >> > > > > > > > > > > such >> > > > > >> >> > > > > > > > > > > > >> > places >> > > > > >> >> > > > > > > > > > > > >> > > > will be quite small. I >> believe >> > > only >> > > > > >> runners >> > > > > >> >> > (but >> > > > > >> >> > > > not >> > > > > >> >> > > > > > > > > > transforms) >> > > > > >> >> > > > > > > > > > > > >> will >> > > > > >> >> > > > > > > > > > > > >> > > ever >> > > > > >> >> > > > > > > > > > > > >> > > > need to set this thread-local >> > > > > >> >> > > > > > > > > > > > >> > > > >> > > > > >> >> > > > > > > > > > > > >> > > > Transform part: >> > > > > >> >> > > > > > > > > > > > >> > > > - Transforms that take >> user-code >> > > > > lambdas >> > > > > >> and >> > > > > >> >> > > want >> > > > > >> >> > > > to >> > > > > >> >> > > > > > let >> > > > > >> >> > > > > > > > > them >> > > > > >> >> > > > > > > > > > > > access >> > > > > >> >> > > > > > > > > > > > >> > side >> > > > > >> >> > > > > > > > > > > > >> > > > inputs still will need to be >> > > > > configurable >> > > > > >> >> > with a >> > > > > >> >> > > > > > method >> > > > > >> >> > > > > > > > like >> > > > > >> >> > > > > > > > > > > > >> > > > .withSideInputs(view1, >> view2...) >> > > and >> > > > > will >> > > > > >> >> need >> > > > > >> >> > > to >> > > > > >> >> > > > > > plumb >> > > > > >> >> > > > > > > > > those >> > > > > >> >> > > > > > > > > > > down >> > > > > >> >> > > > > > > > > > > > >> to >> > > > > >> >> > > > > > > > > > > > >> > the >> > > > > >> >> > > > > > > > > > > > >> > > > primitive DoFn's or >> CombineFn's >> > - >> > > > > >> >> information >> > > > > >> >> > on >> > > > > >> >> > > > > > *which* >> > > > > >> >> > > > > > > > > side >> > > > > >> >> > > > > > > > > > > > inputs >> > > > > >> >> > > > > > > > > > > > >> > will >> > > > > >> >> > > > > > > > > > > > >> > > > be accessed, of course, still >> > > needs >> > > > to >> > > > > >> be in >> > > > > >> >> > the >> > > > > >> >> > > > > > graph. >> > > > > >> >> > > > > > > > > > > > >> > > > >> > > > > >> >> > > > > > > > > > > > >> > > > I quickly prototyped this in >> > > direct >> > > > > >> runner >> > > > > >> >> and >> > > > > >> >> > > > > > > > MapElements, >> > > > > >> >> > > > > > > > > > and >> > > > > >> >> > > > > > > > > > > it >> > > > > >> >> > > > > > > > > > > > >> > worked >> > > > > >> >> > > > > > > > > > > > >> > > > with no issues - I'm >> wondering >> > if >> > > > > there's >> > > > > >> >> > > > something >> > > > > >> >> > > > > > > subtle >> > > > > >> >> > > > > > > > > > that >> > > > > >> >> > > > > > > > > > > > I'm >> > > > > >> >> > > > > > > > > > > > >> > > > missing, or is it actually a >> > good >> > > > > idea? >> > > > > >> >> > > > > > > > > > > > >> > > > >> > > > > >> >> > > > > > > > > > > > >> > > >> > > > > >> >> > > > > > > > > > > > >> > >> > > > > >> >> > > > > > > > > > > > >> >> > > > > >> >> > > > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > > >> > > > > >> >> > > > > > > > > > > >> > > > > >> >> > > > > > > > > > >> > > > > >> >> > > > > > > > > >> > > > > >> >> > > > > > > > >> > > > > >> >> > > > > > > >> > > > > >> >> > > > > > >> > > > > >> >> > > > > >> > > > > >> >> > > > >> > > > > >> >> > > >> > > > > >> >> > >> > > > > >> >> >> > > > > >> >> > > > > >> > > > >> > > >> > >> >
