title: How does Dataflow autoscale and distribute workload?
body: <p>After reading <a href="https://stackoverflow.com/questions/51179443/cloud-dataflow-how-does-dataflow-do-parallelism">this question</a>, I still have some questions about how Dataflow/Apache Beam distribute workload. The problem I'm running into can be demonstrated with the following code:</p>
  
  <pre class="lang-java prettyprint-override"><code>package debug;
  
  import java.io.IOException;
  
  import org.apache.beam.runners.dataflow.DataflowRunner;
  import org.apache.beam.runners.dataflow.options.DataflowPipelineOptions;
  import org.apache.beam.runners.dataflow.options.DataflowPipelineWorkerPoolOptions;
  import org.apache.beam.sdk.Pipeline;
  import org.apache.beam.sdk.PipelineResult;
  import org.apache.beam.sdk.io.gcp.pubsub.PubsubIO;
  import org.apache.beam.sdk.options.PipelineOptionsFactory;
  import org.apache.beam.sdk.transforms.DoFn;
  import org.apache.beam.sdk.transforms.ParDo;
  
  public class DebugPipeline {
      @SuppressWarnings("serial")
      public static PipelineResult main(String[] args) throws IOException {
  
          /*******************************************
           * SETUP - Build options.
           ********************************************/
  
          DataflowPipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation()
                  .as(DataflowPipelineOptions.class);
          options.setRunner(DataflowRunner.class);
          options.setAutoscalingAlgorithm(
                  DataflowPipelineWorkerPoolOptions.AutoscalingAlgorithmType.THROUGHPUT_BASED);
          // Autoscaling will scale between n/15 and n workers, so from 1-15 here
          options.setMaxNumWorkers(15);
          // Default of 250GB is absurdly high and we don't need that much on every worker
          options.setDiskSizeGb(32);
          // Manually configure scaling (i.e. 1 vs 5 for comparison)
          options.setNumWorkers(5);
  
          // Debug Pipeline
          Pipeline pipeline = Pipeline.create(options);
          pipeline
              .apply(PubsubIO.readStrings()
                      .fromSubscription("your subscription"))
              // this is the transform that I actually care about. In production code, this will
              // send a REST request to some 3rd party endpoint.
              .apply("sleep", ParDo.of(new DoFn&lt;String, String&gt;() {
                  @ProcessElement
                  public void processElement(ProcessContext c) throws InterruptedException {
                      Thread.sleep(500);
                      c.output(c.element());
                  }
              }));
  
          return pipeline.run();
      }
  }
  </code></pre>
  
  <p>Comparing the maximum throughput when using 1 worker and 5 workers, instead of the latter perform 5 times more efficiently, it is only marginally more efficiently. It makes me wonder about the following questions:</p>
  
  <ol>
  <li>Assume each worker is using 4 vCPU, is each thread bound to a particular DoFn, or the same DoFns can be invoked on all thread at a given moment if needed to boost performace?</li>
  <li>Assume there are more than one workers, will each worker gets a complete pipeline, i.e. at least an instance of each Transform including the source?</li>
  <li>How do Dataflow/Apache Beam determine which Transform to invoke more often? Will is create more instances of the DoFn that takes up more CPU resources? longer wall time? or every Transform gets copied same mount of time?</li>
  <li>According to the <a href="https://beam.apache.org/documentation/programming-guide/#overview" rel="nofollow noreferrer">Apache programming guide</a>, that the backend is equivalent to <code>asynchronous “job”</code>. Does that mean each DoFn instance is processed asynchronously?</li>
  <li>On the same note, in the example code provided, how would the "sleep" Transform be processed in asynchronously?</li>
  <li>In the production code, <code>Thread.sleep</code> is replaced with a synchronous http request to a 3rd party API. Does the asynchronous process means it will convert the synchronous client into asynchronous?</li>
  </ol>
  
  <p><strong>Update</strong></p>
  
  <p>And an extra question: 
  the <a href="https://cloud.google.com/dataflow/service/dataflow-service-desc#autoscaling" rel="nofollow noreferrer">Dataflow documentation</a> has a comment on PubSubIO that: </p>
  
  <blockquote>
    <p>In extreme cases (i.e. Cloud Pub/Sub subscriptions with large publishing batches or sinks with very high latency), autoscaling is known to become coarse-grained. </p>
  </blockquote>
  
  <p>Could you expand on: </p>
  
  <ol>
  <li><p>What <strong>large publishing batches</strong> mean? i.e. large batch size or large number of batches? </p></li>
  <li><p>Does the <strong>high latency sink</strong> include high latency in transforms before the sink? </p></li>
  <li><p>What is the <strong>coarse-grained</strong> behaviour?</p></li>
  </ol>