Python-如何-大查询异步任务

Question

这可能是个假问题，但我似乎无法异步运行python google clood bigquery。

我的目标是并发运行多个查询，并等待所有查询在asyncio.wait（）query gatherer中完成。我正在使用asyncio.create_tast（）启动查询。问题是，每个查询在开始之前都要等待前一个查询完成。

这里是我的查询功能（相当简单）：

async def exec_query(self, query, **kwargs) -> bigquery.table.RowIterator:
  job = self.api.query(query, **kwargs)
  return job.result()

因为我不能等待job.result（）我应该等待别的东西吗？

Answer 1

事实上，由于使用了asyncio.create\u task（）函数，我找到了一种将查询封装在async调用中的方法，非常容易。我只需要将job.result（）打包到一个协同程序中；下面是实现。它现在确实是异步运行的。

class BQApi(object):                                                                                                 
    def __init__(self):                                                                                              
        self.api = bigquery.Client.from_service_account_json(BQ_CONFIG["credentials"])                               

    async def exec_query(self, query, **kwargs) -> bigquery.table.RowIterator:                                       
        job = self.api.query(query, **kwargs)                                                                        
        task = asyncio.create_task(self.coroutine_job(job))                                                          
        return await task                                                                                            

    @staticmethod                                                                                                    
    async def coroutine_job(job):                                                                                    
        return job.result()                                                                                          

Answer 2

只是为了分享一个不同的解决方案：

import numpy as np
from time import sleep


query1 = """
SELECT
  language.name,
  average(language.bytes)
FROM `bigquery-public-data.github_repos.languages` 
, UNNEST(language) AS language
GROUP BY language.name"""
query2 = 'SELECT 2'


def dummy_callback(future):
    global jobs_done
    jobs_done[future.job_id] = True


jobs = [bq.query(query1), bq.query(query2)]
jobs_done = {job.job_id: False for job in jobs}
[job.add_done_callback(dummy_callback) for job in jobs]

# blocking loop to wait for jobs to finish
while not (np.all(list(jobs_done.values()))):
    print('waiting for jobs to finish ... sleeping for 1s')
    sleep(1)

print('all jobs done, do your stuff')

我宁愿使用bigquery作业本身的内置异步功能，而不是在完成时使用。这也使得我能够将数据管道分解为单独的云函数，而不必在整个管道期间保持主ThreadPoolExecutor活动。顺便说一句，这就是我研究这个问题的原因：我的管道比云函数的最大超时时间9分钟（甚至是云运行的15分钟）还要长。

缺点是我需要跟踪各种函数中的所有作业id，但在配置管道时，通过指定输入和输出，使其形成有向无环图，这相对容易解决。

Answer 3

如果您在coroutine中工作，并且希望在不阻塞事件\u循环的情况下运行不同的查询，那么您可以使用run\u in\u executor函数，该函数基本上在后台线程中运行查询，而不阻塞循环。这里有一个很好的例子说明如何使用它。

确保这正是你所需要的；为在Python API中运行查询而创建的作业已经是异步的，它们仅在调用job.result（）时才会阻塞。这意味着您不需要使用asyncio，除非您在协同程序中。

下面是一个在作业完成后立即检索结果的快速示例：

from concurrent.futures import ThreadPoolExecutor, as_completed
import google.cloud.bigquery as bq


client = bq.Client.from_service_account_json('path/to/key.json')
query1 = 'SELECT 1'
query2 = 'SELECT 2'

threads = []
results = []

executor = ThreadPoolExecutor(5)

for job in [client.query(query1), client.query(query2)]:
    threads.append(executor.submit(job.result))

# Here you can run any code you like. The interpreter is free

for future in as_completed(threads):
    results.append(list(future.result()))

结果将是：

[[Row((2,), {'f0_': 0})], [Row((1,), {'f0_': 0})]]