将 BigQuery 脚本的结果返回到 Python 客户端

Question

从 2019 年秋季开始，BigQuery 支持脚本编写，这非常棒。我不知道BigQuery 的 Python 客户端是否能够利用这个新功能。

例如，运行以下 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 客户端？

Answer 1

它受支持，但您需要考虑以下文档：

\n\n

\n

脚本在 BigQuery 中使用 jobs.insert 执行，与任何其他查询类似，并将多语句脚本指定为查询文本。执行脚本时，会为脚本中的每个语句创建附加作业（ 称为子作业）。您可以通过调用 jobs.list 来枚举脚本的子作业，并传入脚本\xe2\x80\x99s\n 作业 ID 作为 ParentJobId 参数。

\n\n

在脚本上调用 jobs.getQueryResults 时，它将返回脚本中执行的最后一个 SELECT、DML 或 DDL 语句的查询结果， 如果上述语句均未执行，则不会返回任何查询结果。要获取脚本中所有语句的结果，请枚举子作业并对每个子作业调用 jobs.getQueryResults。

\n

\n\n

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

\n\n

\n\n

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

\n\n

\n\n

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

\n\n

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

\n\n

\n\n

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

\n\n

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

\n\n

我们使用该list_jobs方法并检查ID 和语句类型：

\n\n

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

\n\n

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

\n\n

from google.cloud import bigquery\n\nclient = bigquery.Client()\nQUERY = """\nBEGIN\n    CREATE OR REPLACE TEMP TABLE t0 AS\n        SELECT name, bikes_count FROM `bigquery-public-data.london_bicycles.cycle_stations` WHERE bikes_count > 10;\n\n    SELECT SUM(bikes_count) AS total_bikes FROM t0;\n\n    DROP TABLE IF EXISTS t0;\nEND;\n"""\n\nquery_job = client.query(QUERY)\nquery_job.result()\n\nfor job in client.list_jobs(parent_job=query_job.job_id):  # list all child jobs\n    # print("Job ID: {}, Statement Type: {}".format(job.job_id, job.statement_type))\n    if job.statement_type == "SELECT":  # print the desired job output only\n        rows = job.result()\n        for row in rows:\n            print("sum={}".format(row["total_bikes"]))\n

\n\n

输出：

\n\n

sum=6676\n

\n