python - 受 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 变量。
• 将结果转储到临时表中，计算行数，然后追加数据 - 但这似乎效率低下。
• 计算查询前后的记录并记录增量 - 这可能会导致并行运行查询出现问题。

有没有最好的方法的建议？

---

跟进马斯喀特的回答，我认为当查询并行运行时这将不起作用:

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，与query_job.result().total_rows基本相同。第二种方式获取有关项目中数据集的信息。这里重要的是，在检查行数之前，我们需要再次调用 table = client.get_table(table_id) 方法。如果我们不这样做，系统将打印:Loaded 0 rows into table，因为它指的是在运行查询之前指定的行数。

希望以上信息对您有用。