限制光束应用中的一步

Posted 2023-03-31

【中文标题】限制光束应用中的一步

【英文标题】：Throttling a step in beam application

【发布时间】：2019-02-10 12:07:32

【问题描述】：

我在谷歌数据流上使用 python 束，我的管道如下所示：

从文件中读取图片网址 >> 下载图片 >> 处理图片

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

这是一种可以限制步骤的方法吗？每分钟输入或输出。

谢谢。

【问题讨论】：

这是一个有趣的问题。我会尽量在星期一之前回答；）

@Xitrum 你有没有想过这个问题或者确实使用了下面的建议？

@Pablo 你有没有想过这个问题？

【参考方案1】：

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

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()); 或等效项。在我的示例中，作为一个用例，我想从公共文件中读取数据，解析出空行并以最大 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 个元素/秒的速率，如下图所示，并避免在使用多个工作人员时达到配额，即使请求并没有真正分散（您可能会同时收到 8 个请求，然后是 8 秒睡眠等）。 ）。这只是一个测试，可能更好的实现是使用 guava 的rateLimiter。

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

【讨论】：

只是想提一下，今晚我尝试使用 Guava 的速率限制器（通过使用 Spotify 为 scio 核心发布的代码）来执行此操作。它工作得很好，除了在日志中看到一些关于长时间不输出的步骤的警告消息（这是有道理的）。我不确定这是否会导致问题。详情请见***.com/questions/69082498/…。