受 BigQuery 查询影响的行数

Question

我每天运行命令将新记录插入 BigQuery 表中，并希望记录每天插入的记录数。

我创建一个QueryJob包含SELECT查询和destination表的对象。我将 设为write_disposition，WRITE_APPEND以便将新数据附加到表中。

我发现两个选项可以做类似的事情，但都没有达到我想要的效果：

• query_job.num_dml_affected_rows：这只是返回 None，因为查询不使用 DML INSERT，而是附加到目标表。
• query_job.result().total_rows：这将返回表中的总行数，而不是新行数。

我可以想到多种方法来实现所需的结果，但不确定最好的方法是什么：

• 将查询更改为 DML 插入 - 但这意味着创建动态 SQL 而不仅仅是使用目标表是一个 Python 变量。
• 将结果转储到临时表中，计算行数，然后附加数据 - 但这似乎效率低下。
• 计算查询之前和之后的记录并记录增量 - 这可能会导致并行运行查询出现问题。

有什么建议没有最好的方法吗？

---

根据 muscat 的回答，我认为当查询并行运行时这将不起作用：

Get number of rows: 1000 rows
Function to call queries in paralell:

 - Query 1 --> Adds 100 rows --> finishes 3rd --> Counts 1200 rows
 - Query 2 --> Adds 80 rows --> finishes 2nd --> Counts 1100 rows
 - Query 3 --> Adds 20 rows --> finishes 1st --> Counts 1020 rows

因为无法知道这些查询将完成哪个顺序（因为它们都是使用库并行调用的multiprocessing），所以我不确定如何知道每个查询添加了多少行？

---

更新2

示例代码：

    ...

    # We compile a list of which datasets need to be loaded from
    brands = self._bq.select(f"Select brand,  gaDataset From {self.BRAND_DATASET}.{self.BRAND_TABLE}")
    brands = list(brands.iterrows())
    _, brands = zip(*brands)

    # Define the function for parallel population
    def populate_fn(brand):
        return self._populate(brand, self.predicates)

    logging.info("Populating daily stats for brands in parallel")
    error = self._parallel_apply(populate_fn, brands)
    if error is not None:
        return error

def _populate(self, brand, predicates):
    # We can't just call <bq_load_data> because we need to update the predicates for each brand
    predicates.add_predicate('gaDataset', brand['gaDataset'], operator="_")
    query_job = self._load_data(self.table_name, predicates=predicates)
    logging.info(f"Started for {brand['gaDataset']}: {brand['brand']}")

    self._run_query_job(query_job)
    logging.info(f"{brand['gaDataset']}: {brand['brand']} is now populated.")

该_populate功能针对每个品牌并行运行。

predicates只是一个处理如何修改 Jinja 模板化 SQL 的对象，其中包含主对象中的一些常见参数，以及一些品牌特定的参数。

_load_data是一个函数，它实际上使用适当的参数加载 Jinja 模板化 SQL，并构造并返回一个QueryJob对象。

Answer 1

有效且推荐的方法是在运行查询之前和之后对记录进行计数。并行运行查询没有问题，因为我们可以等待查询作业完成后再检查更新的行数。我准备了如何检查新添加的行数的示例：

from google.cloud import bigquery

client = bigquery.Client()

# Define destination table.
table_id = "<PROJECT_ID>.<DATASET>.<TABLE>"

# Inspect the number of rows in the table before running the query.
table = client.get_table(table_id)
num_rows_begin = table.num_rows
print("Number of rows before running the query job: " + str(num_rows_begin))

sql = """
    SELECT word, word_count
    FROM `bigquery-public-data.samples.shakespeare`
    LIMIT 10
"""

job_config = bigquery.QueryJobConfig(destination=table_id, write_disposition="WRITE_APPEND")

# Make an API request.
query_job = client.query(sql, job_config=job_config)

# Wait for the job to complete.  
query_job.result()

# Inspect the number of newly added rows in the table after running the query.
# First way:
num_rows_end = query_job._query_results.total_rows - num_rows_begin
print("Loaded {} rows into {}".format(str(num_rows_end), table_id))

# Second way:
table = client.get_table(table_id)
print("Loaded {} rows into {}".format(table.num_rows - num_rows_begin, table_id))

正如您所看到的，检查新添加的行数的方法很少。第一个指的是query_job:的结果query_job._query_results.total_rows，与what基本相同query_job.result().total_rows。第二种方式获取有关项目中数据集的信息。这里重要的是，我们需要table = client.get_table(table_id)在检查行数之前再次调用方法。如果我们不这样做，系统将打印：Loaded 0 rows into table，因为它指的是运行查询之前指定的行数。

我希望上述信息对您有用。