I don't think this is a Dataflow specific question -- other runners likely perform fusion as it is an important optimization to reduce communication overhead within a pipeline.
For the same reason, I also don't think making this a global option is desirable -- in Spark this would be analogous to making every ParDo run as a separate stage. But as you say there are cases where it is necessary to indicate to the runner that fusion may be undesirable. Some examples: - If a ParDo that is expensive to run is fused with a flaky ParDo, we may need to rerun the expensive operation unnecessarily. In this case being able to annotate the ParDo is "expensive" could be a signal that you wanted the runner to avoid fusion since it would prevent these coupled failures. - If a ParDo that produces wildly varying amounts of elements the dependent parallelism may hinder performance of the pipeline if you fuse downstream operations with it. In this case either indicating this property of the ParDo, implementing a Splittable DoFn, or having a way to say "please redistribute elements in this collection" would be useful. Some of these concepts are explained in the https://beam.apache.org/documentation/execution-model/. It is absolutely problematic for portability that you need to use a GroupByKey as the solution to this variety of problems. By categorizing the different problems it is used for -- which may have different solutions in different runners -- we can better understand how to express those requirements in Beam, enabling you to write a portable pipeline and the runner to do the right things. See discussions of the RequiresDeterministicInput annotation for a similar example, where we're introducing an annotation to indicate we require being able to deterministically replay an input collection, rather than relying on not-actually-required semantics of a GroupByKey. -- Ben On Thu, Dec 7, 2017 at 12:35 PM Jacob Marble <[email protected]> wrote: > Is there a long-term plan for preventing fusion in Dataflow pipelines? > Maybe a simple flag --disableFusion ? > > I have read a few of the discussions in the Beam mailing lists, and I > haven't found any sentiment that something should be changed about > Dataflow, only that Dataflow users should work around this with > sometimes-costly GBKs. > > My example is a streaming pipeline: > PubSub => memcached read => http request => memcached write => GCS write > > The three middle steps are implemented as few lines of readable code. > That's how pipeline steps/transformations should be written, agreed? But, > since they are fused, they are all slow. > > I have tried some hacks successfully. For example, a DoFn handles an > element by (1) start an async network request then (2) output any network > responses responses waiting in a response queue. One DoFn instance can > handle thousands of network requests concurrently. Also, this feels like > I'm using Beam incorrectly. > > Jacob >
