我在谷歌数据流上使用 python 束，我的管道如下所示：
从文件中读取图片网址 >> 下载图片 >> 处理图片
问题是我不能让 下载图像 步长按需要进行扩展，因为我的应用程序可能会被图像服务器阻止。
这是一种可以限制步骤的方法吗？每分钟输入或输出。
谢谢。
【问题讨论】：
标签： python google-cloud-dataflow apache-beam dataflow
一种可能性，也许是幼稚的，是在步骤中引入睡眠。为此，您需要知道可以同时运行的 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 的数量。我想看看是否有其他更好的解决方案。
【讨论】：