限制光束应用的一步

Question

我在 google 数据流上使用 python beam，我的管道如下所示：

从文件中读取图像 url >> 下载图像 >> 处理图像

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

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

谢谢你。

Answer 1

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

\n\n

Integer desired_rate = 1; // QPS limit\n\nif (options.getNumWorkers() == 0) { num_workers = 1; }\nelse { num_workers = options.getNumWorkers(); }\n\nif (options.getWorkerMachineType() != null) { \n    machine_type = options.getWorkerMachineType();\n    num_threads = Integer.parseInt(machine_type.substring(machine_type.lastIndexOf("-") + 1));\n}\nelse { num_threads = 1; }\n\nDouble sleep_time = (double)(num_workers * num_threads) / (double)(desired_rate);\n

\n\n

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

\n\n

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

\n\n

速率限制 Fn 为ThrottledFn，注意该sleep函数：\n

\n\n

static class ThrottledFn extends DoFn<String, String> {\n    @ProcessElement\n    public void processElement(ProcessContext c) throws Exception {\n\n        // Instantiates a client\n        try (LanguageServiceClient language = LanguageServiceClient.create()) {\n\n          // The text to analyze\n          String text = c.element();\n          Document doc = Document.newBuilder()\n              .setContent(text).setType(Type.PLAIN_TEXT).build();\n\n          // Detects the sentiment of the text\n          Sentiment sentiment = language.analyzeSentiment(doc).getDocumentSentiment();                 \n          String nlp_results = String.format("Sentiment: score %s, magnitude %s", sentiment.getScore(), sentiment.getMagnitude());\n\n          TimeUnit.SECONDS.sleep(sleep_time.intValue());\n\n          Log.info(nlp_results);\n          c.output(nlp_results);\n        }\n    }\n}\n

\n\n

这样，我得到了 1 个元素/秒的速率，如下图所示，并避免在使用多个工作线程时达到配额，即使请求没有真正分散（您可能会得到 8 个并发请求，然后是 8 秒睡眠等）。这只是一个测试，可能更好的实现是使用番石榴的rateLimiter。

\n\n

\n\n

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

\n