数据流如何自动扩展和分配工作负载？

Question

读完这个问题后，我仍然对DataFlow/Apache Beam如何分配工作负载有一些疑问。我遇到的问题可以用下面的代码演示：

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<String, String>() {
                @ProcessElement
                public void processElement(ProcessContext c) throws InterruptedException {
                    Thread.sleep(500);
                    c.output(c.element());
                }
            }));

        return pipeline.run();
    }
}

比较使用1个worker和5个worker时的最大吞吐量，而不是后者的效率高5倍，它只是稍微高一点。这让我对以下问题产生了疑问：

1. 假设每个工作线程使用4个vCPU，那么每个线程是否绑定到特定的DoFn，或者如果需要提高性能，可以在给定时刻对所有线程调用相同的DoFns？
2. 假设有多个worker，每个worker是否会获得完整的管道，即每个转换的至少一个实例，包括源？
3. dataflow/apache Beam如何确定要更频繁地调用哪个转换？是否会创建更多占用更多CPU资源的DoFn实例？更长的墙时间？还是每个转换都复制相同的时间？
4. 根据Apache编程指南，后端等同于异步“作业”。这是否意味着每个DoFn实例都是异步处理的？
5. 同样注意，在提供的示例代码中，如何异步处理“睡眠”转换？
6. 在生产代码中，thread.sleep被替换为对第三方API的同步http请求。异步进程是否意味着它将把同步客户机转换为异步？

更新

还有一个额外的问题：数据流文档对PubSubIO有一个评论：

在极端情况下（例如，具有大发布批次的云发布/子订阅或具有非常高延迟的接收器），自动伸缩已知会变得粗粒度。

你能详细说明一下吗：

>

大的出版批量意味着什么？即批量大还是批量多？

高延迟接收器是否包括接收器之前转换中的高延迟？

Answer 1

1. DOFNs可以在给定时刻在所有线程上调用，（请看蓝星）
2. 是的，每个工人将处理一个完整的管道
3. 云数据流服务执行各种优化：融合和组合
4. 那些可以的步骤，没有数据依赖项
5. “Hibernate”可以在不同的工人上同时处理，并按顺序发送到队列中。
6. 取决于数据依赖关系