Hey all, An update: https://github.com/apache/incubator-beam/pull/896 has been merged, laying groundwork and adding support for splittable DoFn to the in-memory runner.
What this PR does: - It defines an API, in full accordance with the proposal discussed on this thread. - It adds a mostly runner-agnostic expansion of the ParDo transform for a splittable DoFn, with one runner-specific primitive transform that needs to be overridden by every runner. - It overrides said transform in the in-memory runner, so this works end-to-end in the in-memory runner. - All this code is covered by tests (unit and integration @RunnableOnService) and appears to work properly in combination with the rest of the Beam model: e.g., inputs to a splittable DoFn can be windowed, and their windows and timestamps are transparently propagated. Caveats: - The API is marked @Experimental, but this is an understatement: it is assumed to be in flux and is not intended to be used yet. Overwhelmingly likely, it *will* change in incompatible ways. DO NOT write pipelines with this transform yet. - It only works in the in-memory runner: the vast majority of code is runner-agnostic, but a central runner-specific primitive transform is only overridden by the in-memory runner. My immediate next plan is to make this work in the Cloud Dataflow streaming runner (since this is the runner I'm most familiar with), in order to get experience with what kind of runner hooks are needed and to put the API in shape for adding hooks for other runners - and then work either myself or with the community on making it work in other runners too. Once all runners sufficiently support a particular subset of features, we can start transitioning some connectors or writing new ones using that subset (I expect that streaming connectors will come first). Additionally, the Python SDK is considering using Splittable DoFn as the *only* API for streaming sources (right now it doesn't have any API for that, so there's no compatibility concerns). No implementation work has happened yet, but it seems like a good idea. On Tue, Aug 30, 2016 at 1:45 AM Aljoscha Krettek <aljos...@apache.org> wrote: > Thanks for the explanation Eugene and JB. > > By the way, I'm not trying to find holes in this, I really like the > feature. I just sometimes wonder how a specific thing might be implemented > with this. > > On Mon, 29 Aug 2016 at 18:00 Eugene Kirpichov <kirpic...@google.com.invalid > > > wrote: > > > Hi Aljoscha, > > > > The watermark reporting is done via > > ProcessContinuation.futureOutputWatermark, at the granularity of > returning > > from individual processElement() calls - you return from the call and > give > > a watermark on your future output. We assume that updating watermark is > > sufficient at a per-bundle level (or, if not, then that you can make > > bundles small enough) cause that's the same level at which state changes, > > timers etc. are committed. > > It can be implemented by setting a per-key watermark hold and updating it > > when each call for this element returns. That's the way it is implemented > > in my current prototype > https://github.com/apache/incubator-beam/pull/896 > > (see > > SplittableParDo.ProcessFn) > > > > On Mon, Aug 29, 2016 at 2:55 AM Aljoscha Krettek <aljos...@apache.org> > > wrote: > > > > > Hi, > > > I have another question about this: currently, unbounded sources have > > > special logic for determining the watermark and the system periodically > > > asks the sources for the current watermark. As I understood it, > > watermarks > > > are only "generated" at the sources. How will this work when sources > are > > > implemented as a combination of DoFns and SplittableDoFns? Will > > > SplittableDoFns be asked for a watermark, does this mean that > watermarks > > > can then be "generated" at any operation? > > > > > > Cheers, > > > Aljoscha > > > > > > On Sun, 21 Aug 2016 at 22:34 Eugene Kirpichov > > <kirpic...@google.com.invalid > > > > > > > wrote: > > > > > > > Hi JB, > > > > > > > > Yes, I'm assuming you're referring to the "magic" part on the > transform > > > > expansion diagram. This is indeed runner-specific, and timers+state > are > > > > likely the simplest way to do this for an SDF that does unbounded > > amount > > > of > > > > work. > > > > > > > > On Sun, Aug 21, 2016 at 12:14 PM Jean-Baptiste Onofré < > j...@nanthrax.net > > > > > > > wrote: > > > > > > > > > Anyway, from a runner perspective, we will have kind of API (part > of > > > the > > > > > Runner API) to "orchestrate" the SDF as we discussed during the > call, > > > > > right ? > > > > > > > > > > Regards > > > > > JB > > > > > > > > > > On 08/21/2016 07:24 PM, Eugene Kirpichov wrote: > > > > > > Hi Aljoscha, > > > > > > This is an excellent question! And the answer is, we don't need > any > > > new > > > > > > concepts like "SDF executor" and can rely on the per-key state > and > > > > timers > > > > > > machinery that already exists in all runners because it's > necessary > > > to > > > > > > implement windowing/triggering properly. > > > > > > > > > > > > Note that this is already somewhat addressed in the previously > > posted > > > > > State > > > > > > and Timers proposal https://s.apache.org/beam-state , under > > "per-key > > > > > > workflows". > > > > > > > > > > > > Think of it this way, using the Kafka example: we'll expand it > > into a > > > > > > transform: > > > > > > > > > > > > (1) ParDo { topic -> (unique key, topic, partition, [0, inf))) > for > > > > > > partition in topic.listPartitions() } > > > > > > (2) GroupByKey > > > > > > (3) ParDo { key, topic, partition, R -> Kafka reader code in the > > > > > > proposal/slides } > > > > > > - R is the OffsetRange restriction which in this case will be > > > always > > > > of > > > > > > the form [startOffset, inf). > > > > > > - there'll be just 1 value per key, but we use GBK to just get > > > access > > > > > to > > > > > > the per-key state/timers machinery. This may be runner-specific; > > > maybe > > > > > some > > > > > > runners don't need a GBK to do that. > > > > > > > > > > > > Now suppose the topic has two partitions, P1 and P2, and they get > > > > > assigned > > > > > > unique keys K1, K2. > > > > > > Then the input to (3) will be a collection of: (K1, topic, P1, > [0, > > > > inf)), > > > > > > (K2, topic, P2, [0, inf)). > > > > > > Suppose we have just 1 worker with just 1 thread. Now, how will > > this > > > > > thread > > > > > > be able to produce elements from both P1 and P2? here's how. > > > > > > > > > > > > The thread will process (K1, topic, P1, [0, inf)), checkpoint > > after a > > > > > > certain time or after a certain number of elements are output > (just > > > > like > > > > > > with the current UnboundedSource reading code) producing a > residual > > > > > > restriction R1' (basically a new start timestamp), put R11 into > the > > > > > per-key > > > > > > state and set a timer T1 to resume. > > > > > > Then it will process (K2, topic, P2, [0, inf)), do the same > > > producing a > > > > > > residual restriction R2' and setting a timer T2 to resume. > > > > > > Then timer T1 will fire in the context of the key K1. The thread > > will > > > > > call > > > > > > processElement again, this time supplying R1' as the restriction; > > the > > > > > > process repeats and after a while it checkpoints and stores R1'' > > into > > > > > state > > > > > > of K1. > > > > > > Then timer T2 will fire in the context of K2, run processElement > > for > > > a > > > > > > while, set a new timer and store R2'' into the state of K2. > > > > > > Etc. > > > > > > If partition 1 goes away, the processElement call will return "do > > not > > > > > > resume", so a timer will not be set and instead the state > > associated > > > > with > > > > > > K1 will be GC'd. > > > > > > > > > > > > So basically it's almost like cooperative thread scheduling: > things > > > run > > > > > for > > > > > > a while, until the runner tells them to checkpoint, then they > set a > > > > timer > > > > > > to resume themselves, and the runner fires the timers, and the > > > process > > > > > > repeats. And, again, this only requires things that runners can > > > already > > > > > do > > > > > > - state and timers, but no new concept of SDF executor (and > > > > consequently > > > > > no > > > > > > necessity to choose/tune how many you need). > > > > > > > > > > > > Makes sense? > > > > > > > > > > > > On Sat, Aug 20, 2016 at 9:46 AM Aljoscha Krettek < > > > aljos...@apache.org> > > > > > > wrote: > > > > > > > > > > > >> Hi, > > > > > >> I have another question that I think wasn't addressed in the > > > meeting. > > > > At > > > > > >> least it wasn't mentioned in the notes. > > > > > >> > > > > > >> In the context of replacing sources by a combination of to SDFs, > > how > > > > do > > > > > you > > > > > >> determine how many "SDF executor" instances you need downstream? > > For > > > > the > > > > > >> sake of argument assume that both SDFs are executed with > > > parallelism 1 > > > > > (or > > > > > >> one per worker). Now, if you have a file source that reads from > a > > > > static > > > > > >> set of files the first SDF would emit the filenames while the > > second > > > > SDF > > > > > >> would receive the filenames and emit their contents. This works > > well > > > > and > > > > > >> the downstream SDF can process one filename after the other. > Now, > > > > think > > > > > of > > > > > >> something like a Kafka source. The first SDF would emit the > > > partitions > > > > > (say > > > > > >> 4 partitions, in this example) and the second SDF would be > > > responsible > > > > > for > > > > > >> reading from a topic and emitting elements. Reading from one > topic > > > > never > > > > > >> finishes so you can't process the topics in series. I think you > > > would > > > > > need > > > > > >> to have 4 downstream "SDF executor" instances. The question now > > is: > > > > how > > > > > do > > > > > >> you determine whether you are in the first or the second > > situation? > > > > > >> > > > > > >> Probably I'm just overlooking something and this is already > dealt > > > with > > > > > >> somewhere... :-) > > > > > >> > > > > > >> Cheers, > > > > > >> Aljoscha > > > > > >> > > > > > >> On Fri, 19 Aug 2016 at 21:02 Ismaël Mejía <ieme...@gmail.com> > > > wrote: > > > > > >> > > > > > >>> Hello, > > > > > >>> > > > > > >>> Thanks for the notes both Dan and Eugene, and for taking the > time > > > to > > > > do > > > > > >> the > > > > > >>> presentation and answer our questions. > > > > > >>> > > > > > >>> I mentioned the ongoing work on dynamic scaling on Flink > because > > I > > > > > >> suppose > > > > > >>> that it will address dynamic rebalancing eventually (there are > > > > multiple > > > > > >>> changes going on for dynamic scaling). > > > > > >>> > > > > > >>> > > > > > >>> > > > > > >> > > > > > > > > > > > > > > > https://docs.google.com/document/d/1G1OS1z3xEBOrYD4wSu-LuBCyPUWyFd9l3T9WyssQ63w/edit#heading=h.2suhu1fjxmp4 > > > > > >>> > > > > > >>> > > > > > https://lists.apache.org/list.html?d...@flink.apache.org:lte=1M:FLIP-8 > > > > > >>> > > > > > >>> Anyway I am far from an expert on flink, but probably the flink > > > guys > > > > > can > > > > > >>> give their opinion about this and refer to a more precise > > document > > > > that > > > > > >> the > > > > > >>> ones I mentioned.. > > > > > >>> > > > > > >>> Thanks again, > > > > > >>> Ismaël > > > > > >>> > > > > > >>> On Fri, Aug 19, 2016 at 8:52 PM, Jean-Baptiste Onofré < > > > > j...@nanthrax.net > > > > > > > > > > > >>> wrote: > > > > > >>> > > > > > >>>> Great summary Eugene and Dan. > > > > > >>>> > > > > > >>>> And thanks again for the details, explanation, and discussion. > > > > > >>>> > > > > > >>>> Regards > > > > > >>>> JB > > > > > >>>> > > > > > >>>> > > > > > >>>> On 08/19/2016 08:16 PM, Eugene Kirpichov wrote: > > > > > >>>> > > > > > >>>>> Thanks for attending, everybody! > > > > > >>>>> > > > > > >>>>> Here are meeting notes (thanks Dan!). > > > > > >>>>> > > > > > >>>>> Q: Will SplittableDoFn enable better repartitioning of the > > > > > >> input/output > > > > > >>>>> data? > > > > > >>>>> A: Not really; repartitioning is orthogonal to SDF. > > > > > >>>>> > > > > > >>>>> Current Source API suffers from lack of composition and > > > scalability > > > > > >>>>> because > > > > > >>>>> we treat sources too much as metadata, not enough as data. > > > > > >>>>> > > > > > >>>>> Q(slide with transform expansion): who does the "magic"? > > > > > >>>>> A: The runner. Checkpointing and dynamically splitting > > > restrictions > > > > > >> will > > > > > >>>>> require collaboration with the runner. > > > > > >>>>> > > > > > >>>>> Q: How does the runner interact with the DoFn to control the > > > > > >>> restrictions? > > > > > >>>>> Is it related to the centralized job tracker etc.? > > > > > >>>>> A: RestrictionTracker is a simple helper object, that exists > > > purely > > > > > on > > > > > >>> the > > > > > >>>>> worker while executing a single partition, and interacts with > > the > > > > > >> worker > > > > > >>>>> harness part of the runner. Not to be confused with the > > > centralized > > > > > >> job > > > > > >>>>> tracker (master) - completely unrelated. Worker harness, of > > > course, > > > > > >>>>> interacts with the master in some relevant ways (e.g. > Dataflow > > > > master > > > > > >>> can > > > > > >>>>> tell "you're a straggler, you should split"). > > > > > >>>>> > > > > > >>>>> Q: Is this a new DoFn subclass, or how will this integrate > with > > > the > > > > > >>>>> existing code? > > > > > >>>>> A: It's a feature of reflection-based DoFn ( > > > > > >>> https://s.apache.org/a-new-do > > > > > >>>>> fn) > > > > > >>>>> - just another optional parameter of type RestrictionTracker > to > > > > > >>>>> processElement() which is dynamically bound via reflection, > so > > > > fully > > > > > >>>>> backward/forward compatible, and looks to users like a > regular > > > > DoFn. > > > > > >>>>> > > > > > >>>>> Q: why is fractionClaimed a double? > > > > > >>>>> A: It's a number in 0..1. Job-wide magic (e.g. autoscaling, > > > dynamic > > > > > >>>>> rebalancing) requires a uniform way to represent progress > > through > > > > > >>>>> different > > > > > >>>>> sources. > > > > > >>>>> > > > > > >>>>> Q: Spark runner is microbatch-based, so this seems to map > well > > > onto > > > > > >>>>> checkpoint/resume, right? > > > > > >>>>> A: Yes; actually the Dataflow runner is, at a worker level, > > also > > > > > >>>>> microbatch-based. The way SDF interacts with a runner will be > > > very > > > > > >>> similar > > > > > >>>>> to how a Bounded/UnboundedSource interacts with a runner. > > > > > >>>>> > > > > > >>>>> Q: Using SDF, what would be the "packaging" of the IO? > > > > > >>>>> A: Same as currently: package IO's as PTransforms and their > > > > > >>> implementation > > > > > >>>>> under the hood can be anything: Source, simple ParDo's, SDF, > > etc. > > > > > E.g. > > > > > >>>>> Datastore was recently refactored from BoundedSource to ParDo > > > > (ended > > > > > >> up > > > > > >>>>> simpler and more scalable), transparently to users. > > > > > >>>>> > > > > > >>>>> Q: What's the timeline; what to do with the IOs currently in > > > > > >>> development? > > > > > >>>>> A: Timeline is O(months). Keep doing what you're doing and > > > working > > > > on > > > > > >>> top > > > > > >>>>> of Source APIs when necessary and simple ParDo's otherwise. > > > > > >>>>> > > > > > >>>>> Q: What's the impact for the runner writers? > > > > > >>>>> A: Tentatively expected that most of the code for running an > > SDF > > > > will > > > > > >> be > > > > > >>>>> common to runners, with some amount of per-runner glue code, > > just > > > > > like > > > > > >>>>> GBK/windowing/triggering. Impact on Dataflow runner is larger > > > since > > > > > it > > > > > >>>>> supports dynamic rebalancing in batch mode and this is the > > > hardest > > > > > >> part, > > > > > >>>>> but for other runners shouldn't be too hard. > > > > > >>>>> > > > > > >>>>> JB: Talend has people who can help with this: e.g. help > > integrate > > > > > into > > > > > >>>>> Spark runner, refactor IOs etc. Amit also willing to chat > about > > > > > >>> supporting > > > > > >>>>> SDF in Spark runner. > > > > > >>>>> > > > > > >>>>> Ismael: There's a Flink proposal about dynamic rebalancing. > > > Ismael > > > > > >> will > > > > > >>>>> send a link. > > > > > >>>>> > > > > > >>>>> > > > > > >>>>> On Fri, Aug 19, 2016 at 7:21 AM Jean-Baptiste Onofré < > > > > > j...@nanthrax.net > > > > > >>> > > > > > >>>>> wrote: > > > > > >>>>> > > > > > >>>>> Hi Eugene, > > > > > >>>>>> > > > > > >>>>>> thanks for the reminder. > > > > > >>>>>> > > > > > >>>>>> Just to prepare some topics for the call, please find some > > > points: > > > > > >>>>>> > > > > > >>>>>> 1. Using SDF, what would be the "packaging" of the IO ? It > > > sounds > > > > to > > > > > >> me > > > > > >>>>>> that we can keep the IO packaging style (using with* setters > > for > > > > the > > > > > >> IO > > > > > >>>>>> configuration) and replace PTransform, Source, Reader, ... > > > > directly > > > > > >>> with > > > > > >>>>>> SDF. Correct ? > > > > > >>>>>> > > > > > >>>>>> 2. What's your plan in term of release to include SDF ? We > > have > > > > > >> several > > > > > >>>>>> IOs in preparation and I wonder if it's worth to start to > use > > > the > > > > > new > > > > > >>>>>> SDF API or not. > > > > > >>>>>> > > > > > >>>>>> 3. What's the impact for the runner writers ? The runners > will > > > > have > > > > > >> to > > > > > >>>>>> support SDF, that could be tricky depending of the execution > > > > engine. > > > > > >> In > > > > > >>>>>> the worst case where the runner can't fully support SDF, > does > > it > > > > > mean > > > > > >>>>>> that most of our IOs will be useless ? > > > > > >>>>>> > > > > > >>>>>> Just my dumb topics ;) > > > > > >>>>>> > > > > > >>>>>> Thanks, > > > > > >>>>>> See you at 8am ! > > > > > >>>>>> > > > > > >>>>>> Regards > > > > > >>>>>> JB > > > > > >>>>>> > > > > > >>>>>> On 08/19/2016 02:29 AM, Eugene Kirpichov wrote: > > > > > >>>>>> > > > > > >>>>>>> Hello everybody, > > > > > >>>>>>> > > > > > >>>>>>> Just a reminder: > > > > > >>>>>>> > > > > > >>>>>>> The meeting is happening tomorrow - Friday Aug 19th, > 8am-9am > > > PST, > > > > > to > > > > > >>>>>>> join > > > > > >>>>>>> the call go to > > > > > >>>>>>> > > > https://hangouts.google.com/hangouts/_/google.com/splittabledofn > > > > . > > > > > >>>>>>> I intend to go over the proposed design and then have a > > > free-form > > > > > >>>>>>> discussion. > > > > > >>>>>>> > > > > > >>>>>>> Please have a skim through the proposal doc: > > > > https://s.apache.org/ > > > > > >>>>>>> splittable-do-fn > > > > > >>>>>>> I also made some slides that are basically a trimmed-down > > > version > > > > > of > > > > > >>> the > > > > > >>>>>>> doc to use as a guide when conducting the meeting, > > > > > >>>>>>> > > > > > >>>>>>> > https://docs.google.com/presentation/d/1DfCp9VRLH-5Ap5nbz047 > > > > > >>>>>> Fiv3NfjrywJVSdef5dRY4aE/edit?usp=sharing > > > > > >>>>>> > > > > > >>>>>>> . > > > > > >>>>>>> > > > > > >>>>>>> I will post notes from the meeting on this thread > afterwards. > > > > > >>>>>>> > > > > > >>>>>>> Thanks, looking forward. > > > > > >>>>>>> > > > > > >>>>>>> On Fri, Aug 12, 2016 at 5:35 PM Dan Halperin > > > > > >>>>>>> <dhalp...@google.com.invalid > > > > > >>>>>>> > > > > > >>>>>>> wrote: > > > > > >>>>>>> > > > > > >>>>>>> This is pretty cool! I'll be there too. (unless the hangout > > > gets > > > > > too > > > > > >>>>>>>> > > > > > >>>>>>> full > > > > > >>>>>> > > > > > >>>>>>> -- if so, I'll drop out in favor of others who aren't lucky > > > > enough > > > > > >> to > > > > > >>>>>>>> > > > > > >>>>>>> get > > > > > >>>>>> > > > > > >>>>>>> to talk to Eugene all the time.) > > > > > >>>>>>>> > > > > > >>>>>>>> On Fri, Aug 12, 2016 at 4:03 PM, Andrew Psaltis < > > > > > >>>>>>>> > > > > > >>>>>>> psaltis.and...@gmail.com> > > > > > >>>>>> > > > > > >>>>>>> wrote: > > > > > >>>>>>>> > > > > > >>>>>>>> +1 I'll join > > > > > >>>>>>>>> > > > > > >>>>>>>>> On Friday, August 12, 2016, Aparup Banerjee (apbanerj) < > > > > > >>>>>>>>> > > > > > >>>>>>>> apban...@cisco.com > > > > > >>>>>>>> > > > > > >>>>>>>>> > > > > > >>>>>>>>>> wrote: > > > > > >>>>>>>>> > > > > > >>>>>>>>> + 1, me2 > > > > > >>>>>>>>>> > > > > > >>>>>>>>>> > > > > > >>>>>>>>>> > > > > > >>>>>>>>>> > > > > > >>>>>>>>>> On 8/12/16, 9:27 AM, "Amit Sela" <amitsel...@gmail.com > > > > > >>>>>>>>>> > > > > > >>>>>>>>> <javascript:;>> > > > > > >>>>>> > > > > > >>>>>>> wrote: > > > > > >>>>>>>>>> > > > > > >>>>>>>>>> +1 as in I'll join ;-) > > > > > >>>>>>>>>>> > > > > > >>>>>>>>>>> On Fri, Aug 12, 2016, 19:14 Eugene Kirpichov > > > > > >>>>>>>>>>> > > > > > >>>>>>>>>> <kirpic...@google.com.invalid > > > > > >>>>>>>>> > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> wrote: > > > > > >>>>>>>>>>> > > > > > >>>>>>>>>>> Sounds good, thanks! > > > > > >>>>>>>>>>>> Then Friday Aug 19th it is, 8am-9am PST, > > > > > >>>>>>>>>>>> > https://staging.talkgadget.google.com/hangouts/_/google > > . > > > > > >>>>>>>>>>>> > > > > > >>>>>>>>>>> com/splittabledofn > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> > > > > > >>>>>>>>>>>> On Thu, Aug 11, 2016 at 11:12 PM Jean-Baptiste Onofré > < > > > > > >>>>>>>>>>>> > > > > > >>>>>>>>>>> j...@nanthrax.net > > > > > >>>>>>>>> > > > > > >>>>>>>>>> <javascript:;>> > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> wrote: > > > > > >>>>>>>>>>>> > > > > > >>>>>>>>>>>> Hi > > > > > >>>>>>>>>>>>> > > > > > >>>>>>>>>>>>> Unfortunately I will be in Ireland on August 15th. > What > > > > about > > > > > >>>>>>>>>>>>> > > > > > >>>>>>>>>>>> Friday > > > > > >>>>>>>> > > > > > >>>>>>>>> 19th ? > > > > > >>>>>>>>>>>> > > > > > >>>>>>>>>>>>> > > > > > >>>>>>>>>>>>> Regards > > > > > >>>>>>>>>>>>> JB > > > > > >>>>>>>>>>>>> > > > > > >>>>>>>>>>>>> > > > > > >>>>>>>>>>>>> > > > > > >>>>>>>>>>>>> On Aug 11, 2016, 23:22, at 23:22, Eugene Kirpichov > > > > > >>>>>>>>>>>>> <kirpic...@google.com.INVALID> wrote: > > > > > >>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>> Hi JB, > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>> Sounds great, does the suggested time over > > > videoconference > > > > > >> work > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>> for > > > > > >>>>>>>> > > > > > >>>>>>>>> you? > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>> On Thu, Aug 11, 2016 at 11:59 AM Jean-Baptiste > Onofré > > < > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>> j...@nanthrax.net <javascript:;>> > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> wrote: > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>> Hi Eugene > > > > > >>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> May we talk together next week ? I like the > > proposal. I > > > > > >> would > > > > > >>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>> just > > > > > >>>>>>>>> > > > > > >>>>>>>>>> need > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> some details for my understanding. > > > > > >>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> Thanks > > > > > >>>>>>>>>>>>>>> Regards > > > > > >>>>>>>>>>>>>>> JB > > > > > >>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> On Aug 11, 2016, 19:46, at 19:46, Eugene Kirpichov > > > > > >>>>>>>>>>>>>>> <kirpic...@google.com.INVALID> wrote: > > > > > >>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> Hi JB, > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> What are your thoughts on this? > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> I'm also thinking of having a virtual meeting to > > > explain > > > > > >> more > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> about > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> this > > > > > >>>>>>>>>>>>>>>> proposal if necessary, since I understand it is a > > lot > > > to > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> digest. > > > > > >>>>>>>> > > > > > >>>>>>>>> > > > > > >>>>>>>>>>>>>>>> How about: Monday Aug 15, 8am-9am Pacific time, > over > > > > > >>> Hangouts? > > > > > >>>>>>>>>>>>>>>> (link: > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>> > > https://staging.talkgadget.google.com/hangouts/_/google. > > > > > >>>>>>>>>>>> > > > > > >>>>>>>>>>> com/splittabledofn > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> - > > > > > >>>>>>>>>>>>>>>> I confirmed that it can be joined without being > > logged > > > > > >> into a > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> Google > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> account) > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> Who'd be interested in attending, and does this > > > > time/date > > > > > >>> work > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> for > > > > > >>>>>>>>> > > > > > >>>>>>>>>> people? > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> On Fri, Aug 5, 2016 at 10:48 AM Eugene Kirpichov > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> <kirpic...@google.com <javascript:;>> > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> wrote: > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> Hi JB, thanks for reading and for your comments! > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> It sounds like you are concerned about continued > > > > support > > > > > >> for > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> existing > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> IO's > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> people have developed, and about backward > > > > compatibility? > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> We do not need to remove the Source API, and all > > > > existing > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> Source-based > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> connectors will continue to work [though the > > document > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> proposes > > > > > >>>>>>>> > > > > > >>>>>>>>> at > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> some > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> point to make Read.from(Source) to translate to a > > > > wrapper > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> SDF > > > > > >>>>>>>> > > > > > >>>>>>>>> under > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> the > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> hood, to exercise the feature more and to make > sure > > > > that > > > > > >> it > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> is > > > > > >>>>>>>> > > > > > >>>>>>>>> strictly > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> more powerful - but this is an optional > > > implementation > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> detail]. > > > > > >>>>>>>>> > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> Perhaps the document phrases this too strongly - > > > > > >> "replacing > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> the > > > > > >>>>>>>>> > > > > > >>>>>>>>>> Source > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> API": a better phrasing would be "introducing a > new > > > API > > > > > so > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> powerful > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> and > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> easy-to-use that hopefully people will choose it > > over > > > > the > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> Source > > > > > >>>>>>>>> > > > > > >>>>>>>>>> API > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> all > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> the time, even though they don't have to" :) And > we > > > can > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> discuss > > > > > >>>>>>>>> > > > > > >>>>>>>>>> whether or > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> not to actually deprecate/remove the Source API > at > > > some > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> point > > > > > >>>>>>>> > > > > > >>>>>>>>> down > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> the > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> road, once it becomes clear whether this is the > > case > > > or > > > > > >> not. > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> To give more context: this proposal came out of > > > > > >> discussions > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> within > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> the SDK > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> team over the past ~1.5 years, before the Beam > > > project > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> existed, > > > > > >>>>>>>>> > > > > > >>>>>>>>>> on > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> how to > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> make major improvements to the Source API; > perhaps > > it > > > > > will > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> clarify > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> things > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> if I give a history of the ideas discussed: > > > > > >>>>>>>>>>>>>>>>> - The first idea was to introduce a > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> Read.from(PCollection<Source>) > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> transform while keeping the Source API intact - this, > > given > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> appropriate > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> implementation, would solve most of the > scalability > > > and > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> composability > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> issues of IO's. Then most connectors would look > like > > : > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> ParDo<A, > > > > > >>>>>>>>> > > > > > >>>>>>>>>> Source<B>> > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> + Read.from(). > > > > > >>>>>>>>>>>>>>>>> - Then we figured that the Source class is an > > > > unnecessary > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> abstraction, as > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> it simply holds data. What if we only had a > > Reader<S, > > > > B> > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> class > > > > > >>>>>>>> > > > > > >>>>>>>>> where > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> S is > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> the source type and B the output type? Then > > > connectors > > > > > >> would > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> be > > > > > >>>>>>>>> > > > > > >>>>>>>>>> something > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> like: ParDo<A, S> + hypothetical > > Read.using(Reader<S, > > > > > B>). > > > > > >>>>>>>>>>>>>>>>> - Then somebody remarked that some of the > features > > of > > > > > >> Source > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> are > > > > > >>>>>>>>> > > > > > >>>>>>>>>> useful to > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> ParDo's as well: e.g. ability to report progress > > when > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> processing a > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> very > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> heavy element, or ability to produce very large > > > output > > > > in > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> parallel. > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> - The two previous bullets were already hinting > that > > > the > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> Read.using() > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> primitive might not be so special: it just takes S > > and > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> produces > > > > > >>>>>>>>> > > > > > >>>>>>>>>> B: > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> isn't > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> that what a ParDo does, plus some source magic, > > minus > > > > the > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> convenience > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> of > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> c.output() vs. the start/advance() state machine? > > > > > >>>>>>>>>>>>>>>>> - At this point it became clear that we should > > > explore > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> unifying > > > > > >>>>>>>>> > > > > > >>>>>>>>>> sources > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> and ParDo's, in particular: can we bring the > magic > > of > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> sources > > > > > >>>>>>>> > > > > > >>>>>>>>> to > > > > > >>>>>>>>> > > > > > >>>>>>>>>> ParDo's > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> but without the limitations and coding > > > inconveniences? > > > > > And > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> this > > > > > >>>>>>>>> > > > > > >>>>>>>>>> is > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> how > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> SplittableDoFn was born: bringing source magic > to a > > > > DoFn > > > > > >> by > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> providing > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> a > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> RangeTracker. > > > > > >>>>>>>>>>>>>>>>> - Once the idea of "splittable DoFn's" was born, > it > > > > > became > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> clear > > > > > >>>>>>>>> > > > > > >>>>>>>>>> that > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> it > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> is strictly more general than sources; at least, > in > > > the > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> respect > > > > > >>>>>>>>> > > > > > >>>>>>>>>> that > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> sources have to produce output, while DoFn's don't: > > an > > > > SDF > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> may > > > > > >>>>>>>> > > > > > >>>>>>>>> very > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> well > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> produce no output at all, and simply perform a > side > > > > > effect > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> in > > > > > >>>>>>>> > > > > > >>>>>>>>> a > > > > > >>>>>>>>> > > > > > >>>>>>>>>> parallel/resumable way. > > > > > >>>>>>>>>>>>>>>>> - Then there were countless hours of discussions > on > > > > > >> unifying > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> the > > > > > >>>>>>>>> > > > > > >>>>>>>>>> bounded/unbounded cases, on the particulars of > > RangeTracker > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> APIs > > > > > >>>>>>>>> > > > > > >>>>>>>>>> reconciling parallelization and checkpointing, what the > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> relation > > > > > >>>>>>>>> > > > > > >>>>>>>>>> between > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> SDF and DF should be, etc. They culminated in the > > > > current > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> proposal. > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> The > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> proposal comes at a time when a couple of key > > > > ingredients > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> are > > > > > >>>>>>>> > > > > > >>>>>>>>> (almost) > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> ready: NewDoFn to make SDF look like a regular > > DoFn, > > > > and > > > > > >> the > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> State/Timers > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> proposal to enable unbounded work per element. > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> To put it shortly: > > > > > >>>>>>>>>>>>>>>>> - Yes, we will support existing Source > connectors, > > > and > > > > > >> will > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> support > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> writing new ones, possibly forever. There is no > > > > > interference > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> with > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> current > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> users of Source. > > > > > >>>>>>>>>>>>>>>>> - The new API is an attempt to improve the Source > > > API, > > > > > >> taken > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> to > > > > > >>>>>>>>> > > > > > >>>>>>>>>> its > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> logical limit where it turns out that users' goals > > can > > > be > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> accomplished > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> easier and more generically entirely within > > ParDo's. > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> Let me know what you think, and thanks again! > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> On Fri, Aug 5, 2016 at 2:39 AM Jean-Baptiste > Onofré > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>> <j...@nanthrax.net <javascript:;>> > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> wrote: > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> Hi Eugene, > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> Just a question: why is it in DoFn and note an > > > > > >> improvement > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> of > > > > > >>>>>>>> > > > > > >>>>>>>>> Source > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> ? > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> If I understand correctly, it means that we will > > > have > > > > to > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> refactore > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> all > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> existing IO: basically, what you propose is to > > remove > > > > all > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> Source > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> to > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> replace with NewDoFn. > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> I'm concern with this approach, especially in > term > > > of > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> timing: > > > > > >>>>>>>> > > > > > >>>>>>>>> clearly, > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> the IO is the area where we have to move forward > in > > > > Beam > > > > > >> as > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> it > > > > > >>>>>>>>> > > > > > >>>>>>>>>> will > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> allow new users to start in their projects. > > > > > >>>>>>>>>>>>>>>>>> So, we started to bring new IOs: Kafka, JMS, > > > > Cassandra, > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> MongoDB, > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> JDBC, > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> ... and some people started to learn the IO API > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> (Bounded/Unbouded > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> source, etc). > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> I think it would make more sense to enhance the > IO > > > API > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> (Source) > > > > > >>>>>>>>> > > > > > >>>>>>>>>> instead > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> of introducing a NewDoFn. > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> What are your thoughts for IO writer like me ? > ;) > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> Regards > > > > > >>>>>>>>>>>>>>>>>> JB > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> On 08/04/2016 07:45 PM, Eugene Kirpichov wrote: > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> Hello Beam community, > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> We (myself, Daniel Mills and Robert Bradshaw) > > would > > > > > like > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> to > > > > > >>>>>>>> > > > > > >>>>>>>>> propose > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> "Splittable DoFn" - a major generalization of > DoFn, > > > > which > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> allows > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> processing > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> of a single element to be non-monolithic, i.e. > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> checkpointable > > > > > >>>>>>>>> > > > > > >>>>>>>>>> and > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> parallelizable, as well as doing an unbounded > amount > > of > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> work > > > > > >>>>>>>>> > > > > > >>>>>>>>>> per > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> element. > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> This allows effectively replacing the current > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> Bounded/UnboundedSource > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> APIs > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> with DoFn's that are much easier to code, more > > > > scalable > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> and > > > > > >>>>>>>> > > > > > >>>>>>>>> composable > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> with > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> the rest of the Beam programming model, and > > enables > > > > > many > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> use > > > > > >>>>>>>>> > > > > > >>>>>>>>>> cases > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> that > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> were previously difficult or impossible, as well > as > > > > some > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> non-obvious new > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> use cases. > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> This proposal has been mentioned before in JIRA > > > > > >> [BEAM-65] > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> and > > > > > >>>>>>>>> > > > > > >>>>>>>>>> some > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> Beam > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> meetings, and now the whole thing is written up > in > > a > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> document: > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> https://s.apache.org/splittable-do-fn > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> Here are some things that become possible with > > > > > >> Splittable > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> DoFn: > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> - Efficiently read a filepattern matching millions of > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> files > > > > > >>>>>>>> > > > > > >>>>>>>>> - Read a collection of files that are produced by an > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> earlier > > > > > >>>>>>>>> > > > > > >>>>>>>>>> step > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> in the > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> pipeline (e.g. easily implement a connector to a > > > > storage > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> system > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> that can > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> export itself to files) > > > > > >>>>>>>>>>>>>>>>>>> - Implement a Kafka reader by composing a "list > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> partitions" > > > > > >>>>>>>> > > > > > >>>>>>>>> DoFn > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> with a > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> DoFn that simply polls a consumer and outputs new > > > > records > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> in > > > > > >>>>>>>>> > > > > > >>>>>>>>>> a > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> while() > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> loop > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> - Implement a log tailer by composing a DoFn > that > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> incrementally > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> returns > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> new > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> files in a directory and a DoFn that tails a > file > > > > > >>>>>>>>>>>>>>>>>>> - Implement a parallel "count friends in > common" > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> algorithm > > > > > >>>>>>>> > > > > > >>>>>>>>> (matrix > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> squaring) with good work balancing > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> Here is the meaningful part of a hypothetical > > Kafka > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> reader > > > > > >>>>>>>> > > > > > >>>>>>>>> written > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> against > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> this API: > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> ProcessContinuation processElement( > > > > > >>>>>>>>>>>>>>>>>>> ProcessContext context, > > > > OffsetRangeTracker > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> tracker) > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> { > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> try (KafkaConsumer<String, String> consumer = > > > > > >>>>>>>>>>>>>>>>>>> > > > > > Kafka.subscribe(context.element().topic, > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> context.element().partition)) { > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> consumer.seek(tracker.start()); > > > > > >>>>>>>>>>>>>>>>>>> while (true) { > > > > > >>>>>>>>>>>>>>>>>>> ConsumerRecords<String, String> > > records = > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> consumer.poll(100ms); > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> if (records == null) return done(); > > > > > >>>>>>>>>>>>>>>>>>> for (ConsumerRecord<String, String> > > > record > > > > : > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> records) > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> { > > > > > >>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>> if > (!tracker.tryClaim(record.offset())) { > > > > > >>>>>>>>>>>>>>>>>>> return > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> > > > > resume().withFutureOutputWatermark(record.timestamp()); > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> } > > > > > >>>>>>>>>>>>>>>>>>> context.output(record); > > > > > >>>>>>>>>>>>>>>>>>> } > > > > > >>>>>>>>>>>>>>>>>>> } > > > > > >>>>>>>>>>>>>>>>>>> } > > > > > >>>>>>>>>>>>>>>>>>> } > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> The document describes in detail the > motivations > > > > behind > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> this > > > > > >>>>>>>>> > > > > > >>>>>>>>>> feature, > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> the > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> basic idea and API, open questions, and > outlines > > an > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> incremental > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>> delivery > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> plan. > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> The proposed API builds on the reflection-based > > new > > > > > DoFn > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> [new-do-fn] > > > > > >>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>> and is > > > > > >>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>>> loosely related to "State and Timers for DoFn" > > > > > >>>>>>>>>>>>>>>>>>> > > > > > >>>>>>>>>>>>>>>>>> [beam-state]. > > > > > >>>>>>>>> > > > > > >>>>>>>>>> > > > > > >>>>>>>>>>
