Apache分束DoFn不会在数据流上将工作分散到多个工作人员

Question

我正在学习使用可拆分DOFN。我预计我的工作将分配给500名员工，但Dataflow只运行了1或2名员工。我是否错误地理解或实现了可拆分DoFn？

我的beam版本是2.16.0

    class Calculate extends DoFn<String, String> {

    private static final long serialVersionUID = 1L;

    @ProcessElement
    public void process(ProcessContext c, RestrictionTracker<OffsetRange, Long> tracker) {
        for (long i = tracker.currentRestriction().getFrom(); tracker.tryClaim(i); ++i) {
            try {
                c.output(i + "_" + InetAddress.getLocalHost().getHostName() + "_" + Math.random() + "_" +  c.element());
                for (int j = 0; j < 10000000; j++ ) {
                    Math.sqrt(j);
                }
            } catch (UnknownHostException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
    @GetInitialRestriction
    public OffsetRange getInitialRange(String element) {
        return new OffsetRange(1L, 50000000L);
    }
    @SplitRestriction
    public void splitRestriction(String element, OffsetRange restriction, OutputReceiver<OffsetRange> receiver) {

        for (long i = restriction.getFrom(); i < restriction.getTo(); i += 100000) {
            receiver.output(new OffsetRange(i, i + 100000));
        }

    }
   }

public static void main(String[] args) {

    //PipelineOptions options = PipelineOptionsFactory.create();
    PipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().as(PipelineOptions.class);
    Pipeline p = Pipeline.create(options);

    p.apply(TextIO.read().from("gs://mybucket/EMEA/20200525/singleposition.csv"))
     .apply(ParDo.of(new Calculate()))
     .apply(TextIO.write().to("gs://mybucket/EMEA/20200525/obligor.csv"));

    p.run().waitUntilFinish();
  }

Answer 1

我在你的理解中发现了两个主要的差距。

1. 如何控制云数据流中的工人数量（例如，您希望为您的Calculate（）DoFn（）生成500个工人）？

云数据流提供2种缩放算法。

b)None(--autoscaling_algorithm=None):在此中，您可以手动提供希望为您的作业生成的员工数量。您可以使用--num_worker参数指定员工数量。

在您的用例中，您使用的是默认模式。因此，您只能得到2个工人取决于您的负荷。

请阅读此链接以获得更详细的信息。

-读取kafka主题的所有分区列表，并从每个分区读取数据（有界源）

上面的用例在使用ParDo（阅读旧ParDo）实现时非常棘手，因为

ParDo fucknitions是单块的--单个processElement（）调用，只能输出有限数量的元素，但不能提供进度或大小调整信息，不能接受并行化提示、拆分或检查点。

现在，在您的场景中，我看不到任何非单片处理的需求。在这里，使用SDF不会获得任何额外的性能好处。