python - 节流光束应用中的一个步骤

Question

我在 google dataflow 上使用 python beam，我的管道如下所示:

Read image urls from file >> Download images >> Process images

问题是我不能让下载图像按需要的步长缩放，因为我的应用程序可能会被图像服务器阻止。

我可以通过这种方式来限制步骤吗？每分钟的输入或输出。

谢谢。

Answer 1

一种可能性，也许是天真的，是在步骤中引休眠眠。为此，您需要知道可以同时运行的 ParDo 实例的最大数量。如果 autoscalingAlgorithm 设置为 NONE，您可以从 numWorkers 和 workerMachineType (DataflowPipelineOptions) 获取它。准确地说，有效率将除以线程总数:desired_rate/(num_workers*num_threads(per worker))。 sleep 时间将是该有效速率的倒数:

Integer desired_rate = 1; // QPS limit

if (options.getNumWorkers() == 0) { num_workers = 1; }
else { num_workers = options.getNumWorkers(); }

if (options.getWorkerMachineType() != null) { 
    machine_type = options.getWorkerMachineType();
    num_threads = Integer.parseInt(machine_type.substring(machine_type.lastIndexOf("-") + 1));
}
else { num_threads = 1; }

Double sleep_time = (double)(num_workers * num_threads) / (double)(desired_rate);

然后您可以在节流的 Fn 中使用 TimeUnit.SECONDS.sleep(sleep_time.intValue()); 或等效项。在我的例子中，作为一个用例，我想从一个公共(public)文件中读取，解析出空行并以最大 1 QPS 的速率调用自然语言处理 API(我将 desired_rate 初始化为 1以前):

p
    .apply("Read Lines", TextIO.read().from("gs://apache-beam-samples/shakespeare/kinglear.txt"))
    .apply("Omit Empty Lines", ParDo.of(new OmitEmptyLines()))
    .apply("NLP requests", ParDo.of(new ThrottledFn()))
    .apply("Write Lines", TextIO.write().to(options.getOutput()));

限速Fn为ThrottledFn，注意sleep函数:

static class ThrottledFn extends DoFn<String, String> {
    @ProcessElement
    public void processElement(ProcessContext c) throws Exception {

        // Instantiates a client
        try (LanguageServiceClient language = LanguageServiceClient.create()) {

          // The text to analyze
          String text = c.element();
          Document doc = Document.newBuilder()
              .setContent(text).setType(Type.PLAIN_TEXT).build();

          // Detects the sentiment of the text
          Sentiment sentiment = language.analyzeSentiment(doc).getDocumentSentiment();                 
          String nlp_results = String.format("Sentiment: score %s, magnitude %s", sentiment.getScore(), sentiment.getMagnitude());

          TimeUnit.SECONDS.sleep(sleep_time.intValue());

          Log.info(nlp_results);
          c.output(nlp_results);
        }
    }
}

有了这个，我得到了 1 element/s 的速率，如下图所示，并且在使用多个 worker 时避免达到配额，即使请求并没有真正分散(你可能会同时收到 8 个请求，然后 8s sleep ，等等。 ).这只是一个测试，可能更好的实现是使用 Guava 的 rateLimiter .

如果管道使用自动缩放 (THROUGHPUT_BASED)，那么它会更复杂，并且应该更新工作人员的数量(例如，Stackdriver Monitoring 有一个 job/current_num_vcpus 指标)。其他一般考虑因素是通过使用虚拟 GroupByKey 或使用 splitIntoBundles 等拆分源来控制并行 ParDos 的数量。我想看看是否有其他更好的解决方案。