Here are some figures (small warn is I did priviledge beam a lot in this benchmark, a bit more than it should in a real impl, I'll say more about it after):
I copied the code at: https://gist.github.com/rmannibucau/fd98fb6a10f9557613fb145c8e7e2de1 And results at: https://gist.github.com/rmannibucau/3d3d0d61e85d45f2959900f208e71c5e Summary is: Benchmark Mode Cnt Score Error Units Comparison.beam thrpt 5 217003888,048 ± 4951801,578 ops/s Comparison.beamCompletionStage thrpt 5 55437442,544 ± 1750845,680 ops/s Comparison.beamFastCompletionStage thrpt 5 128422225,642 ± 3215651,832 ops/s Comparison.defaultCompletionStage thrpt 5 57284761,644 ± 1022305,051 ops/s Comparison.fastCompletionStage thrpt 5 202739102,801 ± 5384078,170 ops/s Comparison.passthrough thrpt 5 391503537,773 ± 6147672,843 ops/s Comments: 1. completionstage then*(Function|Consumer) overhead is visible and creating more objects than beam it is a bit slower 2. it is trivial to make completionstage faster than the default in all the "precomputed" cases beam can know about (we mentionned it earlier) and it falls in the "fastCompletionStage" benchmarks which are x2 adapted to beam case (and allows to dropp one instantiation and compare and swap logic) 3. the diff between the fast completionstage and beam is mainly about the creation of one more wrapper + lambda handling 4. however figures stays high (milllions ops/s!). if you add any IO or real computation into the mix you will converge to a sensitively close number 5. don't forget that being based on CompletionStage we can become reactive/async and scale more appropriately if runners embrace that and at the end be faster anyway (the old NIO vs BIO topic ;)) To summarize: I don't think these figures means that completionstage is a bad fit for beam but actually the opposite even if the raw difference is significative. It still shows that the completionstage based throughoutput is high enough to fit real world batches IMHO. Just to illustrate that point, if I just do a System.getProperty("user.name").length() (my username being rmannibucau): @Benchmark public void stringLength(final Blackhole blackhole) { blackhole.consume(System.getProperty("user.name").length()); } the execution is slower than completionfuture in beam case: Comparison.stringLength thrpt 5 42428432,508 ± 4555866,693 ops/s Romain Manni-Bucau @rmannibucau <https://twitter.com/rmannibucau> | Blog <https://rmannibucau.metawerx.net/> | Old Blog <http://rmannibucau.wordpress.com/> | Github <https://github.com/rmannibucau> | LinkedIn <https://www.linkedin.com/in/rmannibucau> | Book <https://www.packtpub.com/application-development/java-ee-8-high-performance> 2018-03-12 18:58 GMT+01:00 Romain Manni-Bucau <rmannibu...@gmail.com>: > > No more but can try to gather some figures and compare it to beam dofn overhead which should be at the same level or a bit more here since it is never unwrapped whereas completionfuture is a raw code chain without beam in the middle. > > Le 12 mars 2018 18:18, "Lukasz Cwik" <lc...@google.com> a écrit : >> >> Do you have data that supports this? >> >> Note that in reality for something like passing an element between DoFns, the constant in o(1) actually matters. Decreasing SDK harness overhead is a good thing though. >> >> On Mon, Mar 12, 2018 at 10:14 AM, Romain Manni-Bucau < rmannibu...@gmail.com> wrote: >>> >>> By itself just the overhead of instantiating a wrapper (so nothing with the recent JVM GC improvement done for the stream/optional usages). After if you use the chaining you have a light overhead but still o(1) you can desire to skip when doing sync code but which will enable you to run way faster doing IO/async code by optimizing the CPU usage properly when you tune your slaves/workers. So tempted to summarize it as "has an overhead allowing to not run slower". It doesn't prevent beam to still expose a synchronous API collapse at evaluation time in a single fn which will give you the best of both worlds. >>> >>> >>> Romain Manni-Bucau >>> @rmannibucau | Blog | Old Blog | Github | LinkedIn | Book >>> >>> 2018-03-12 18:08 GMT+01:00 Lukasz Cwik <lc...@google.com>: >>>> >>>> It is expected that SDKs will have all their cores fully utilized by processing bundles in parallel and not by performing intrabundle parallelization. This allows for DoFns to be chained together via regular method calls because the overhead to pass a single element through all the DoFn's should be as minimal as possible >>>> >>>> What is the overhead of using a completion stage vs using a regular method call? >>>> >>>> On Sun, Mar 11, 2018 at 10:18 PM, Romain Manni-Bucau < rmannibu...@gmail.com> wrote: >>>>> >>>>> >>>>> >>>>> Le 12 mars 2018 00:19, "Reuven Lax" <re...@google.com> a écrit : >>>>> >>>>> >>>>> >>>>> >>>>> On Sun, Mar 11, 2018 at 7:40 AM Romain Manni-Bucau < rmannibu...@gmail.com> wrote: >>>>>> >>>>>> makes me think beam should maybe do 2 internals changes before moving forward on (s)df API changes: >>>>>> >>>>>> 1. define a beam singleton per JVM (classloader hierarchy actually but you get the idea I think) which can be used to store things locally for reuse - see 2 for an example or metrics pusher work Etienne does could benefit from it too >>>>> >>>>> >>>>> I think we do need something like this, but it needs to be a bit more than a singleton per JVM. For one thing it needs to be at least per pipeline within a JVM. You might run multiple tests in a single JVM, and it should also be possible to run those tests in parallel without the static state interfering with each other. I also think the state needs to be addressable per step (i.e. a ParDo can look up its static state without caring about static state belonging to another ParDo). >>>>> >>>>> >>>>> Agree but you can register in a singleton the pipeline (as "ref" not as instance) and therefore hit the same need. +1 to have scopes (singleton, pipeline, thread) but it still requires a single singleton to handle serialization ;). >>>>> >>>>> >>>>> >>>>> >>>>>> 2. define a SPI to load (s)dofn parameter provider instead of having an ArgProvider which provides everything which is supported. This way you can use any kind of parameter and the parameterproviders can use 1. to handle their own state. First impl of the parameterprovider SPI would be a) state b) timer c) reactive handlers and potentially user parameter providers (service like which can be singleton in the scope of a "JVM" thanks to 1). >>>>>> >>>>>> >>>>>> Romain Manni-Bucau >>>>>> @rmannibucau | Blog | Old Blog | Github | LinkedIn | Book >>>>>> >>>>>> 2018-03-11 15:32 GMT+01:00 Reuven Lax <re...@google.com>: >>>>>>> >>>>>>> Yep. Introduce OutputEmitter, and Process context no longer has much use. >>>>>>> >>>>>>> On Sun, Mar 11, 2018, 11:19 AM Romain Manni-Bucau < rmannibu...@gmail.com> wrote: >>>>>>>> >>>>>>>> Which is still a key feature for sdf but agree it can be dropped for an outputemitter pattern and the dofn moved to a plain parameters injection based pattern. Both (which completionstage) stays compatible :). >>>>>>>> >>>>>>>> Le 11 mars 2018 13:12, "Reuven Lax" <re...@google.com> a écrit : >>>>>>>>> >>>>>>>>> I think process context should go away completely. At that point it has little use except for a way to send output downstream. >>>>>>>>> >>>>>>>>> On Sun, Mar 11, 2018, 6:07 AM Romain Manni-Bucau < rmannibu...@gmail.com> wrote: >>>>>>>>>> >>>>>>>>>> Hmm, thinking out loud but completionstage should/could be extended to replace processcontext since it represents element and output at the same time no? >>>>>>>>>> >>>>>>>>>> Le 11 mars 2018 00:57, "Kenneth Knowles" <k...@google.com> a écrit : >>>>>>>>>>> >>>>>>>>>>> Yea, I think it could. But it is probably more readable to not overload the term, plus certainly a bit simpler in implementation. So perhaps @AsyncElement to make it very clear. >>>>>>>>>>> >>>>>>>>>>> Kenn >>>>>>>>>>> >>>>>>>>>>> On Sat, Mar 10, 2018 at 1:32 PM Reuven Lax <re...@google.com> wrote: >>>>>>>>>>>> >>>>>>>>>>>> Ken, can NewDoFn distinguish at generation time the difference between: >>>>>>>>>>>> >>>>>>>>>>>> public void process(@Element CompletionStage<InputT> element, ...) { >>>>>>>>>>>> >>>>>>>>>>>> and >>>>>>>>>>>> >>>>>>>>>>>> public void process(@Element Input element, ...) { >>>>>>>>>>>> >>>>>>>>>>>> If not, then we would probably need separate annotations.... >>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>>> On Sat, Mar 10, 2018 at 11:09 AM Kenneth Knowles < k...@google.com> wrote: >>>>>>>>>>>>> >>>>>>>>>>>>> Nice! I agree that providing a CompletionStage for chaining is much better than an ExecutorService, and very clear. >>>>>>>>>>>>> >>>>>>>>>>>>> It is very feasible to add support that looks like >>>>>>>>>>>>> >>>>>>>>>>>>> new DoFn<InputT, OutputT>() { >>>>>>>>>>>>> @ProcessElement >>>>>>>>>>>>> public void process(@Element CompletionStage<InputT> element, ...) { >>>>>>>>>>>>> element.thenApply(...) >>>>>>>>>>>>> } >>>>>>>>>>>>> } >>>>>>>>>>>>> >>>>>>>>>>>>> If we had this available, I think users could even experiment with this often as it might help even where it isn't obvious. >>>>>>>>>>>>> >>>>>>>>>>>>> My main hesitation is that big part of Beam is giving a basic/imperative style of programming a DoFn that executes in a very smart functional/parallel way. Full future-oriented programming is not explored much outside of Javascript (and maybe Haskell) and requires greater discipline in programming in a functional manner - if you are mutating stuff in your callback you are going to have bugs, and then when you add concurrency control you are going to have bad performance and deadlocks. So I definitely wouldn't make it the default or want to spend all our support effort on teaching advanced programming technique. >>>>>>>>>>>>> >>>>>>>>>>>>> Kenn >>>>>>>>>>>>> >>>>>>>>>>>>> On Sat, Mar 10, 2018 at 9:31 AM Romain Manni-Bucau < rmannibu...@gmail.com> wrote: >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> 2018-03-10 17:30 GMT+01:00 Reuven Lax <re...@google.com>: >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> Have you considered drafting in detail what you think this API might look like? >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> Yes, but it is after the "enhancements" - for my use cases - and "bugs" list so didn't started to work on it much. >>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> If it's a radically different API, it might be more appropriate as an alternative parallel Beam API rather than a replacement for the current API (there is also one such fluent API in the works). >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> What I plan is to draft it on top of beam (so the "useless" case I spoke about before) and then propose to impl it ~natively and move it as main API for another major. >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> On Sat, Mar 10, 2018 at 7:23 AM Romain Manni-Bucau < rmannibu...@gmail.com> wrote: >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> 2018-03-10 16:19 GMT+01:00 Reuven Lax <re...@google.com>: >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> This is another version (maybe a better, Java 8 idiomatic one?) of what Kenn suggested. >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> Note that with NewDoFn this need not be incompatible (so might not require waiting till Beam 3.0). We can recognize new parameters to processElement and populate add needed. >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> This is right however in my head it was a single way movemenent to enforce the design to be reactive and not fake a reactive API with a sync and not reactive impl which is what would be done today with both support I fear. >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> On Sat, Mar 10, 2018, 12:13 PM Romain Manni-Bucau < rmannibu...@gmail.com> wrote: >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> Yes, for the dofn for instance, instead of having processcontext.element()=<T> you get a CompletionStage<T> and output gets it as well. >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> This way you register an execution chain. Mixed with streams you get a big data java 8/9/10 API which enabkes any connectivity in a wel performing manner ;). >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> Le 10 mars 2018 13:56, "Reuven Lax" <re...@google.com> a écrit : >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> So you mean the user should have a way of registering asynchronous activity with a callback (the callback must be registered with Beam, because Beam needs to know not to mark the element as done until all associated callbacks have completed). I think that's basically what Kenn was suggesting, unless I'm missing something. >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> On Fri, Mar 9, 2018 at 11:07 PM Romain Manni-Bucau < rmannibu...@gmail.com> wrote: >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> Yes, callback based. Beam today is synchronous and until bundles+combines are reactive friendly, beam will be synchronous whatever other parts do. Becoming reactive will enable to manage the threading issues properly and to have better scalability on the overall execution when remote IO are involved. >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> However it requires to break source, sdf design to use completionstage - or equivalent - to chain the processing properly and in an unified fashion. >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> Le 9 mars 2018 23:48, "Reuven Lax" <re...@google.com> a écrit : >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> If you're talking about reactive programming, at a certain level beam is already reactive. Are you referring to a specific way of writing the code? >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> On Fri, Mar 9, 2018 at 1:59 PM Reuven Lax < re...@google.com> wrote: >>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>> What do you mean by reactive? >>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>> On Fri, Mar 9, 2018, 6:58 PM Romain Manni-Bucau < rmannibu...@gmail.com> wrote: >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> @Kenn: why not preferring to make beam reactive? Would alow to scale way more without having to hardly synchronize multithreading. Elegant and efficient :). Beam 3? >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> Le 9 mars 2018 22:49, "Kenneth Knowles" < k...@google.com> a écrit : >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>> I will start with the "exciting futuristic" answer, which is that we envision the new DoFn to be able to provide an automatic ExecutorService parameters that you can use as you wish. >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>> new DoFn<>() { >>>>>>>>>>>>>>>>>>>>>>> @ProcessElement >>>>>>>>>>>>>>>>>>>>>>> public void process(ProcessContext ctx, ExecutorService executorService) { >>>>>>>>>>>>>>>>>>>>>>> ... launch some futures, put them in instance vars ... >>>>>>>>>>>>>>>>>>>>>>> } >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>> @FinishBundle >>>>>>>>>>>>>>>>>>>>>>> public void finish(...) { >>>>>>>>>>>>>>>>>>>>>>> ... block on futures, output results if appropriate ... >>>>>>>>>>>>>>>>>>>>>>> } >>>>>>>>>>>>>>>>>>>>>>> } >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>> This way, the Java SDK harness can use its overarching knowledge of what is going on in a computation to, for example, share a thread pool between different bits. This was one reason to delete IntraBundleParallelization - it didn't allow the runner and user code to properly manage how many things were going on concurrently. And mostly the runner should own parallelizing to max out cores and what user code needs is asynchrony hooks that can interact with that. However, this feature is not thoroughly considered. TBD how much the harness itself manages blocking on outstanding requests versus it being your responsibility in FinishBundle, etc. >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>> I haven't explored rolling your own here, if you are willing to do the knob tuning to get the threading acceptable for your particular use case. Perhaps someone else can weigh in. >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>> Kenn >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>> On Fri, Mar 9, 2018 at 1:38 PM Josh Ferge < josh.fe...@bounceexchange.com> wrote: >>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>> Hello all: >>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>> Our team has a pipeline that make external network calls. These pipelines are currently super slow, and the hypothesis is that they are slow because we are not threading for our network calls. The github issue below provides some discussion around this: >>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>> https://github.com/apache/beam/pull/957 >>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>> In beam 1.0, there was IntraBundleParallelization, which helped with this. However, this was removed because it didn't comply with a few BEAM paradigms. >>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>> Questions going forward: >>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>> What is advised for jobs that make blocking network calls? It seems bundling the elements into groups of size X prior to passing to the DoFn, and managing the threading within the function might work. thoughts? >>>>>>>>>>>>>>>>>>>>>>>> Are these types of jobs even suitable for beam? >>>>>>>>>>>>>>>>>>>>>>>> Are there any plans to develop features that help with this? >>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>> Thanks >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>> >>>>> >>>> >>> >>
