将 BigQuery 脚本的结果返回给 Python 客户端答案

【问题标题】：Returning Results of BigQuery Script to Python Client将 BigQuery 脚本的结果返回给 Python 客户端
【发布时间】：2020-01-19 01:07:55
【问题描述】：

截至 2019 年秋季，BigQuery 支持 scripting，这非常棒。我不知道Python client for BigQuery 是否能够利用这个新功能。

例如，运行以下 Python 代码：

client = bigquery.Client()
QUERY = """
BEGIN
    CREATE OR REPLACE TEMP TABLE t0 AS
        SELECT * FROM my_dataset.my_table WHERE foo < 1;

    SELECT SUM(bar) AS bar_sum FROM t0;

    DROP TABLE IF EXISTS t0;
END;
"""

query_job = client.query(QUERY)
rows = query_job.result()

...返回 google.cloud.bigquery.table._EmptyRowIterator 对象，即使我能够看到 SQL 脚本中的语句已从 BigQuery 的 Web UI 成功运行。

如何将标准 SQL 脚本中 SELECT 语句的结果返回给 Python 客户端？

【问题讨论】：

标签： python python-3.x google-bigquery

【解决方案1】：

支持，但您需要考虑以下documentation：

脚本在 BigQuery 中使用 jobs.insert 执行，类似于任何其他查询，将多语句脚本指定为查询文本。当脚本执行时，额外的作业，称为子作业， 为脚本中的每个语句创建。你可以枚举通过调用jobs.list，传入脚本的子作业作业 ID 作为 parentJobId 参数。

在脚本上调用 jobs.getQueryResults 时，它将返回 要执行的最后一个 SELECT、DML 或 DDL 语句的查询结果 脚本，如果以上语句都没有，则没有查询结果执行。要获取脚本中所有语句的结果，枚举子作业并在每个作业上调用jobs.getQueryResults 他们。

例如，我修改了您的脚本以查询公共表：bigquery-public-data.london_bicycles.cycle_stations。这会运行三个子作业：

最后一个删除表格并且不返回任何行：

这就是为什么，如果我运行 Python 文件，我会得到类似 <google.cloud.bigquery.table._EmptyRowIterator object at 0x7f440aa33c88> 的信息。

我们想要的是中间查询的输出结果：

快速测试是注释掉DROP 语句，然后遍历行以获得sum=6676 的结果。那么，如果我们想要中间结果呢？与前面引用的文档一样，答案是调用 jobs.list 并将脚本作业 ID 作为 parentJobId 参数传递以获取子作业 ID：

for job in client.list_jobs(parent_job=query_job.job_id):
    print("Job ID: {}, Statement Type: {}".format(job.job_id, job.statement_type))

我们使用list_jobs方法并检查ID and statement type：

Job ID: script_job_80e...296_2, Statement Type: DROP_TABLE
Job ID: script_job_9a0...7fd_1, Statement Type: SELECT
Job ID: script_job_113...e13_0, Statement Type: CREATE_TABLE_AS_SELECT

请注意，后缀 (0, 1, 2) 表示执行顺序，但我们可以在检索结果之前添加双重检查以验证作业实际上是 SELECT 语句：

from google.cloud import bigquery

client = bigquery.Client()
QUERY = """
BEGIN
    CREATE OR REPLACE TEMP TABLE t0 AS
        SELECT name, bikes_count FROM `bigquery-public-data.london_bicycles.cycle_stations` WHERE bikes_count > 10;

    SELECT SUM(bikes_count) AS total_bikes FROM t0;

    DROP TABLE IF EXISTS t0;
END;
"""

query_job = client.query(QUERY)
query_job.result()

for job in client.list_jobs(parent_job=query_job.job_id):  # list all child jobs
    # print("Job ID: {}, Statement Type: {}".format(job.job_id, job.statement_type))
    if job.statement_type == "SELECT":  # print the desired job output only
        rows = job.result()
        for row in rows:
            print("sum={}".format(row["total_bikes"]))

输出：

sum=6676

【讨论】：