python - 云数据流流式传输，空闲时停止以省钱？

Question

我有一个应用程序，用户可以在其中投票。

我希望扩大我的应用程序的规模，因此我决定使用 Cloud Dataflow 聚合 Firestore 中存储的计数器。

我已经设置了一个流式类型的 Dataflow 作业，因此只要用户对某项内容进行投票，它就可以监听 pubsub 主题。

有时我每天有数千个用户输入，有时我有几百个...当工作一段时间没有收到 pubsub 消息时，有什么解决方案可以“暂停”该工作吗？

目前，我的数据流作业始终在运行，恐怕这会花费我很多钱。

如果有人可以帮助我了解流媒体工作的计费，我们将不胜感激

这是我的 Python 管道:

def run(argv=None):
    # Config
    parser = argparse.ArgumentParser()
    # Output PubSub Topic
    parser.add_argument(
        '--output_topic', required=True)
    # Input PubSub Topic
    parser.add_argument(
        '--input_topic', required=True)

    known_args, pipeline_args = parser.parse_known_args(argv)

    # Pipeline options
    pipeline_options = PipelineOptions(pipeline_args)
    pipeline_options.view_as(SetupOptions).save_main_session = True
    pipeline_options.view_as(StandardOptions).streaming = True

    # Pipeline process
    with beam.Pipeline(options=pipeline_options) as p:

        # Counting votes
        def count_votes(contestant_votes):
            (contestant, votes) = contestant_votes
            return (contestant, sum(votes))

        # Format data to a fake object (used to be parsed by the CF)
        def format_result(contestant_votes):
            (contestant, votes) = contestant_votes
            return '{ "contestant": %s, "votes": %d }' % (contestant, votes)

        transformed = (p
                       | 'Receive PubSub' >> beam.io.ReadFromPubSub(topic=known_args.input_topic)
                       .with_output_types(bytes)
                       | 'Decode' >> beam.Map(lambda x: x.decode('utf-8'))
                       | 'Pair with one' >> beam.Map(lambda x: (x, 1))
                       | 'Apply window of time' >> beam.WindowInto(window.FixedWindows(30, 0))
                       | 'Group by contestant' >> beam.GroupByKey()
                       | 'Count votes' >> beam.Map(count_votes)
                       | 'Format to fake object string' >> beam.Map(format_result)
                       | 'Transform to PubSub base64 string' >> beam.Map(lambda x: x.encode('utf-8'))
                       .with_output_types(bytes))

        # Trigger a the output PubSub topic with the message payload
        transformed | beam.io.WriteToPubSub(known_args.output_topic)

        result = p.run()
        result.wait_until_finish()


if __name__ == '__main__':
    logging.getLogger().setLevel(logging.INFO)
    run()

Answer 1

回答您的成本问题:对于您当前使用的工作器，您将花费大约 250 美元(取决于您当月的 PD 使用情况)。

目前无需等待即可强制数据流“空闲”或扩展到 0 个工作线程。您可以拥有的最小值是 1。

话虽这么说，您可以采取几条路线来尽量降低成本。

如果您的工作线程负载不大，并且您想要最简单的选择，则可以使用功能较弱的工作线程(n1-standard-1 [~USD $77.06] 或 n1-standard-2 [~USD $137.17]) 。 https://cloud.google.com/products/calculator/#id=3bbedf2f-8bfb-41db-9923-d3a5ef0c0250 (如果您看到我添加了所有 3 个版本，使用的是我在您的照片中看到的 430GB PD)。

如果您需要计算能力，您可以切换到使用基于 cron 的数据流作业，如下所述:https://cloud.google.com/blog/products/gcp/scheduling-dataflow-pipelines-using-app-engine-cron-service-or-cloud-functions 。这样，您可能应该从订阅而不是主题中阅读，这样您就可以保留这些消息，直到您开始工作为止。