如何从云运行调用相同的云运行以并行运行请求？

Question

我正在使用 cloud run 运行 ETL 流程。

我有 2000 个文件。只有 1200 个文件在 BIG Query 中得到预处理和加载。因为云运行正在超时。所以，我想到了分担。

我将 2000 个文件分成 4 个，每个 500 个，并验证和使用 requests.post 来调用相同的云运行。但是，它使用相同的云运行实例依次执行一组。然后又超时了

如何让它并行运行？

截至目前，最大实例数：20。并发数：1，CPU：2，内存：8GB。

Answer 1

嗯，我做过类似的事情。我不确定它是否对您有帮助，因为您没有共享一个代码块。这是我要下载的 2k JSON 文件的示例代码。

您有 2000 个文件，其中 1200 个在云运行超时之前在 GBQ 中得到处理/加载。你可以做的是：

    total_files = len(file_list)//1000   #let file list be 2000, total files will be 2.

    #divide the files into sets of 1000 and loop over them one by one
    
    for file_set in range(1,(total_files+1)):
        auth_and_trigger(file_list[(file_set-1)*1000:(file_set*1000)])

    #for files left after 1000*i , we finally trigger it.
    auth_and_trigger(file_list[(total_files)*1000:len(file_list)])

现在，您可以使用 auth 和 trigger 函数为每 1000 个文件调用云运行。

    def auth_and_trigger(self, rest_of_files):
    #your cloud run url
    receiving_service_url = 'https://cloudrun-url-uc.a.run.app/download'

    # Set up metadata server request
    # See https://cloud.google.com/compute/docs/instances/verifying-instance-identity#request_signature
    metadata_server_token_url = 'http://metadata/computeMetadata/v1/instance/service-accounts/default/identity?audience='

    token_request_url = metadata_server_token_url + receiving_service_url
    token_request_headers = {'Metadata-Flavor': 'Google'}

    # Fetch the token
    token_response = requests.get(token_request_url, headers=token_request_headers)
    jwt = token_response.content.decode("utf-8")

    # Provide the token in the request to the receiving service
    receiving_service_headers = {'Authorization': f'bearer {jwt}'}

    try:
        threading.Thread(target=self.trigger_ingest,
                         args=(receiving_service_url,
                               {"files": rest_of_files},
                               receiving_service_headers
                               )).start()
    except Exception as error:
        logging.error(error)

每个线程都会调用一个函数 trigger_ingest 来调用云运行。它的代码如下：

    def trigger_ingest(url, json, headers=""):
    service_response = requests.post(url=url,
                                     json=json,
                                     headers=headers
                                     )
    logging.info(service_response.content)

现在，由于您想要并行执行，因此请确保线程中没有代码重复，因为您在云运行的触发器中拥有它。