我在 google dataflow 上使用 python beam,我的管道如下所示:
Read image urls from file >> Download images >> Process images
问题是我不能让下载图像按需要的步长缩放,因为我的应用程序可能会被图像服务器阻止。
我可以通过这种方式来限制步骤吗?每分钟的输入或输出。
谢谢。
最佳答案
一种可能性,也许是天真的,是在步骤中引休眠眠。为此,您需要知道可以同时运行的 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 的数量。我想看看是否有其他更好的解决方案。
关于python - 节流光束应用中的一个步骤,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/52183538/