如何使用 BigQuery 客户端访问子作业的 DML 结果？答案

【问题标题】：How to access the DML result of the child job with BigQuery Client?如何使用 BigQuery 客户端访问子作业的 DML 结果？
【发布时间】：2021-06-18 21:26:42
【问题描述】：

我想使用 BigQuery 客户端检查 dml_affected_rows 的值。

我已检查 [1]，当查询（或 DML）包含单个作业时它可以工作。但是，当结果是像下面的脚本一样从儿童的作业中写入时，num_dml_affected_rows 是无效的。如何查看子作业的num_dml_affected_rows。我的想法如下。

query_job.children_job[-1].num_dml_affected_rows #（最后一个子 dml 结果）

但是从文档 [2, 3] 中找不到正确的方法（或方法）来完成它。

[1]BigQuery update how to get number of updated rows
[2]BQ REST
[3]BQ Client API (dev)

示例脚本 (python3)

sql = """
    INSERT `project.dataset.table` (logdatetime)
    SELECT
        CURRENT_DATETIME() as logdatetime
    ;
    SELECT 1
    ;
    INSERT `project.dataset.table` (logdatetime)
    SELECT
        CURRENT_DATETIME() as logdatetime
    ;
"""
query_job = bigquery.Client().query(sql)
query_job.result()
print(query_job.num_child_jobs)  # 3
print(query_job.num_dml_affected_rows)  # None

附言我目前的解决方案是在 DML 之前和之后存储 destination table 的行数以检查差异，但如果我能够直接访问子作业的 DML 结果，则这是不必要的。

【问题讨论】：

标签： google-bigquery

【解决方案1】：

脚本的工作方式是system variable@@row_count。

如果在脚本中使用并且之前的脚本语句是 DML，则指定由于该 DML 语句而修改、插入或删除的行数。

在你的情况下，如果你这样做

sql = """
    INSERT `project.dataset.table` (logdatetime)
    SELECT
        CURRENT_DATETIME() as logdatetime
    ;
    SELECT @@row_count -- output how many row inserted by previous DML
    ;
    INSERT `project.dataset.table` (logdatetime)
    SELECT
        CURRENT_DATETIME() as logdatetime
    ;
    SELECT @@row_count -- output how many row inserted by previous DML
    ;
"""
query_job = bigquery.Client().query(sql)
query_job.result()
print(query_job.num_child_jobs)  # you're getting 4 jobs here
print(query_job.num_dml_affected_rows)  # still None

【讨论】：

【解决方案2】：

对于最初的问题，这可能有点晚了，但如果其他人偶然发现这个问题，这是我的解决方案：

def run_bigquery_query():
    bigquery_client = bigquery.Client()

    sql_query = """ SELECT / INSERT / ... """

    job = bigquery_client.query(sql_query)
    job.result()  # Wait for the whole script to finish

    affected_rows = 0
    for child_job in bigquery_client.list_jobs(parent_job=job.job_id):
        if child_job.num_dml_affected_rows is not None:
            affected_rows += child_job.num_dml_affected_rows

    print("Affected rows: {}".format(affected_rows))

【讨论】：